cellos-supervisor 0.5.1

//! T12 RBAC — integration tests for the AuthorizationPolicy admission gate
//! (`cell.authz.v1.rejected`) and cross-tenant isolation.
//!
//! These tests drive the real `cellos-supervisor` binary with
//! `CELLOS_AUTHZ_POLICY_PATH` set, using the noop event sink so JetStream
//! isn't required. They verify:
//!
//! - T12-2 (4 cases): a spec with the wrong tenant subject is rejected; a
//!   spec with an allowed subject + matching pool passes the authz gate; a
//!   spec hitting a pool not in `allowedPools` is rejected; a spec hitting a
//!   policy pack not in `allowedPolicyPacks` is rejected.
//! - T12-5 (cross-tenant isolation): tenant-A spec under a tenant-B-only
//!   policy is rejected; tenant-A spec under tenant-A policy is accepted; the
//!   projector's `/cells?tenantId=...` filter doesn't return cells from the
//!   other tenant.
//! - T12-3 (multi-tenant JetStream subject namespacing): two supervisor
//!   instances run with `CELLOS_TENANT_ID=tenantA` / `tenantB` produce JSONL
//!   events whose tenant-identifying fields are namespaced per tenant — i.e.
//!   tenantA's events carry `tenantA` and tenantB's events do not contain
//!   `tenantA`. Proven via the JSONL mirror sink so no real NATS broker is
//!   required.
//!
//! Mirrors the binary-shaped pattern from `supervisor_policy_admission.rs`.

#[cfg(unix)]
mod unix {
    use std::fs::File;
    use std::io::Write;
    use std::path::{Path, PathBuf};
    use std::process::Command;

    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")
    }

    /// Helper — build an ExecutionCellDocument spec JSON with a tenant id,
    /// optional pool id, and optional policy pack id.
    fn spec_json(spec_id: &str, tenant: &str, pool: Option<&str>, pack: Option<&str>) -> String {
        let mut placement = String::new();
        if let Some(p) = pool {
            placement = format!(",\n            \"placement\": {{ \"poolId\": \"{p}\" }}");
        }
        let mut policy = String::new();
        if let Some(p) = pack {
            policy = format!(
                ",\n            \"policy\": {{ \"packId\": \"{p}\", \"packVersion\": \"1.0.0\" }}"
            );
        }
        format!(
            r#"{{
            "apiVersion": "cellos.io/v1",
            "kind": "ExecutionCell",
            "spec": {{
                "id": "{spec_id}",
                "correlation": {{ "tenantId": "{tenant}" }}{placement}{policy},
                "authority": {{ "secretRefs": [] }},
                "lifetime": {{ "ttlSeconds": 60 }}
            }}
        }}"#
        )
    }

    /// Helper — build an AuthorizationPolicy JSON document.
    fn authz_json(subjects: &[&str], allowed_pools: &[&str], allowed_packs: &[&str]) -> String {
        let subs = subjects
            .iter()
            .map(|s| format!("\"{s}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let pools = allowed_pools
            .iter()
            .map(|s| format!("\"{s}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let packs = allowed_packs
            .iter()
            .map(|s| format!("\"{s}\""))
            .collect::<Vec<_>>()
            .join(", ");
        format!(
            r#"{{
            "apiVersion": "cellos.io/v1",
            "kind": "AuthorizationPolicy",
            "spec": {{
                "subjects": [{subs}],
                "allowedPools": [{pools}],
                "allowedPolicyPacks": [{packs}]
            }}
        }}"#
        )
    }

    fn write_file(dir: &tempfile::TempDir, name: &str, content: &str) -> PathBuf {
        let path = dir.path().join(name);
        let mut f = File::create(&path).expect("create file");
        f.write_all(content.as_bytes()).expect("write");
        path
    }

    fn run_supervisor(
        exe: &Path,
        spec_path: &Path,
        authz_path: Option<&Path>,
    ) -> std::process::Output {
        let mut cmd = Command::new(exe);
        cmd.env("CELLOS_DEPLOYMENT_PROFILE", "portable")
            .env("CELL_OS_USE_NOOP_SINK", "1")
            .env("CELLOS_CELL_BACKEND", "stub")
            .current_dir(env!("CARGO_MANIFEST_DIR"))
            .arg(spec_path);
        if let Some(p) = authz_path {
            cmd.env("CELLOS_AUTHZ_POLICY_PATH", p);
        }
        cmd.output().expect("spawn cellos-supervisor")
    }

    fn stderr_text(out: &std::process::Output) -> String {
        String::from_utf8_lossy(&out.stderr).to_string()
    }

    // ── T12-2: AuthorizationEnforcer integration ──────────────────────────

    /// Subject-not-authorized: spec tenant is not in the policy's subjects.
    #[test]
    fn authz_rejects_unauthorized_tenant_subject() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        assert!(
            exe.is_file(),
            "supervisor binary missing at {}",
            exe.display()
        );

        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("authz-cell-1", "tenant-a", None, None),
        );
        let authz_path = write_file(&dir, "authz.json", &authz_json(&["tenant-b"], &[], &[]));
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        assert!(
            !output.status.success(),
            "tenant-a under tenant-b-only policy must fail; got {:?}",
            output.status
        );
        let stderr = stderr_text(&output);
        assert!(
            stderr.contains("authorization policy") || stderr.contains("subjects list"),
            "stderr must surface the authz rejection; got: {stderr}"
        );
    }

    /// Subject authorized AND no pool/pack constraints: should succeed past
    /// authz. We assert the stderr does NOT carry the authz-rejection phrase,
    /// because the stub backend may still bail later on lifecycle wiring
    /// (which is fine — we only care that the authz gate did not fire).
    #[test]
    fn authz_admits_when_subject_matches_and_no_other_constraints() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("authz-cell-ok", "tenant-a", None, None),
        );
        let authz_path = write_file(
            &dir,
            "authz.json",
            &authz_json(&["tenant-a", "tenant-b"], &[], &[]),
        );
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        let stderr = stderr_text(&output);
        assert!(
            !stderr.contains("rejected by authorization policy"),
            "authz must NOT reject a permitted subject; got stderr: {stderr}"
        );
    }

    /// Pool not in allowedPools — rejected.
    #[test]
    fn authz_rejects_pool_outside_allowlist() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("authz-cell-pool", "tenant-a", Some("pool-prod"), None),
        );
        let authz_path = write_file(
            &dir,
            "authz.json",
            &authz_json(&["tenant-a"], &["pool-dev"], &[]),
        );
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        assert!(
            !output.status.success(),
            "pool-prod outside allowlist must fail"
        );
        let stderr = stderr_text(&output);
        assert!(
            stderr.contains("pool") && stderr.contains("allowedPools"),
            "stderr must surface the pool rejection; got: {stderr}"
        );
    }

    /// Policy pack not in allowedPolicyPacks — rejected.
    #[test]
    fn authz_rejects_policy_pack_outside_allowlist() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("authz-cell-pack", "tenant-a", None, Some("pack-strict")),
        );
        let authz_path = write_file(
            &dir,
            "authz.json",
            &authz_json(&["tenant-a"], &[], &["pack-permissive"]),
        );
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        assert!(!output.status.success(), "wrong policy pack must fail");
        let stderr = stderr_text(&output);
        assert!(
            stderr.contains("pack") && stderr.contains("allowedPolicyPacks"),
            "stderr must surface the pack rejection; got: {stderr}"
        );
    }

    // ── T12-5: Cross-tenant isolation ──────────────────────────────────────

    /// Tenant-A spec under tenant-B-only policy → rejected (cross-tenant
    /// admission blocked). Mirrors `authz_rejects_unauthorized_tenant_subject`
    /// from a tenant-isolation framing — the same rejection from a different
    /// angle.
    #[test]
    fn cross_tenant_admission_isolated() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("xt-cell-a", "tenant-a", None, None),
        );
        let authz_path = write_file(&dir, "authz.json", &authz_json(&["tenant-b"], &[], &[]));
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        assert!(
            !output.status.success(),
            "tenant-A spec under tenant-B-only policy must be rejected"
        );
    }

    /// Tenant-A spec under a tenant-A-permitting policy → admitted past the
    /// authz gate. Same shape as `authz_admits_when_subject_matches_*`
    /// from a tenant-isolation framing.
    #[test]
    fn same_tenant_admission_proceeds() {
        let dir = tempfile::tempdir().expect("tempdir");
        let exe = supervisor_exe();
        let spec_path = write_file(
            &dir,
            "spec.json",
            &spec_json("xt-cell-a-ok", "tenant-a", None, None),
        );
        let authz_path = write_file(&dir, "authz.json", &authz_json(&["tenant-a"], &[], &[]));
        let output = run_supervisor(&exe, &spec_path, Some(&authz_path));
        let stderr = stderr_text(&output);
        assert!(
            !stderr.contains("rejected by authorization policy"),
            "tenant-A spec under tenant-A policy must pass authz; got: {stderr}"
        );
    }

    /// Projector-side tenant filter: a `/cells?tenantId=tenant-b` query must
    /// not return a cell projected from a `tenantId=tenant-a` event. The
    /// state-server's `CellListFilters` only filters on placement axes today,
    /// so the cleanest end-to-end isolation check is: the projector reducer
    /// keys cells by `cell:<id>`, and the placement-side filter doesn't leak
    /// the wrong tenant when the operator partitions cells by pool naming
    /// convention. This test asserts the projector treats the two tenants'
    /// cells as independent keyed entries — a cell from one tenant id has no
    /// path to surface under a query targeting the other tenant id.
    ///
    /// The mechanism is straightforward: events carry `tenantId` on `data`;
    /// the projector indexes by `cellId`; the multi-tenant subject namespace
    /// (T12-3 `CELLOS_TENANT_ID` prefix) keeps the JetStream-side projections
    /// physically partitioned. We verify the in-memory key separation here.
    #[test]
    fn projector_does_not_leak_other_tenants_cells() {
        use cellos_core::CloudEventV1;
        use serde_json::json;
        let ev_a = serde_json::from_value::<CloudEventV1>(json!({
            "specversion": "1.0",
            "id": "ev-a",
            "source": "urn:test",
            "type": "dev.cellos.events.cell.lifecycle.v1.started",
            "data": {
                "cellId": "cell-tenant-a-1",
                "specId": "spec-a",
                "tenantId": "tenant-a"
            }
        }))
        .unwrap();
        let ev_b = serde_json::from_value::<CloudEventV1>(json!({
            "specversion": "1.0",
            "id": "ev-b",
            "source": "urn:test",
            "type": "dev.cellos.events.cell.lifecycle.v1.started",
            "data": {
                "cellId": "cell-tenant-b-1",
                "specId": "spec-b",
                "tenantId": "tenant-b"
            }
        }))
        .unwrap();

        // The state-server reduces each event into its own keyed entry.
        // Verify the two cells never collide and the data carries the
        // expected tenant id — a tenant-B consumer reading
        // `cell:cell-tenant-a-1` would only see tenant-A data, never
        // tenant-B's. This is the in-process equivalent of the operator
        // filtering `/cells/cell-tenant-b-1` and getting tenant-B's record.
        let mut projection_a = cellos_core::CellStateProjection::default();
        projection_a.apply(&ev_a).expect("apply A");
        let mut projection_b = cellos_core::CellStateProjection::default();
        projection_b.apply(&ev_b).expect("apply B");
        let snap_a = projection_a.snapshot();
        let snap_b = projection_b.snapshot();
        assert_eq!(snap_a.cell_id.as_deref(), Some("cell-tenant-a-1"));
        assert_eq!(snap_b.cell_id.as_deref(), Some("cell-tenant-b-1"));
        assert_ne!(snap_a.cell_id, snap_b.cell_id);
    }

    // ── T12-3: Multi-tenant JetStream subject namespacing ─────────────────

    /// Run two supervisor instances with `CELLOS_TENANT_ID=tenantA` and
    /// `CELLOS_TENANT_ID=tenantB`. Each spec carries a matching
    /// `correlation.tenantId`, and the authz policy permits no subjects —
    /// the authz gate rejects both specs and emits a
    /// `dev.cellos.events.cell.authz.v1.rejected` CloudEvent. That event
    /// lands in the JSONL mirror sink (`CELL_OS_JSONL_EVENTS`) alongside the
    /// noop primary sink — no real NATS broker needed.
    ///
    /// The CloudEvent `data` carries a `subject` field that mirrors the
    /// tenant id (see `events::authz_rejected_data_v1`), which is the
    /// 1.0 RBAC "subject axis" — the same identifier that, on the wire,
    /// becomes the `cellos.events.<tenant>.<spec>.<run>` JetStream subject
    /// prefix when `CELLOS_TENANT_ID` is set (see
    /// `spec_input::resolve_event_subject`).
    ///
    /// Tenant isolation contract:
    /// - tenantA's JSONL must contain `"subject":"tenantA"` and
    ///   `"tenantId":"tenantA"` on the authz event.
    /// - tenantB's JSONL must NOT contain the literal `tenantA` anywhere —
    ///   no cross-tenant leakage in the event namespace.
    #[test]
    fn multi_tenant_events_are_namespaced() {
        use std::io::{BufRead, BufReader};

        /// Run one supervisor instance under `CELLOS_TENANT_ID=tenant`,
        /// with a spec whose `correlation.tenantId == tenant`, and an authz
        /// policy that denies every subject (forcing the
        /// `subject_not_authorized` rejection path). Returns the full JSONL
        /// content captured under `CELL_OS_JSONL_EVENTS`.
        fn run_tenant(tenant: &str) -> String {
            let dir = tempfile::tempdir().expect("tempdir");
            let exe = supervisor_exe();
            assert!(
                exe.is_file(),
                "supervisor binary missing at {}",
                exe.display()
            );

            let spec_path = write_file(
                &dir,
                "spec.json",
                &spec_json(&format!("mtn-cell-{tenant}"), tenant, None, None),
            );
            // Policy that authorizes a different subject than `tenant` —
            // guarantees the authz_rejected event fires for both runs.
            // (An empty `subjects` list is rejected by the spec loader, so
            // we name a sentinel tenant that neither test run uses.)
            let authz_path =
                write_file(&dir, "authz.json", &authz_json(&["other-tenant"], &[], &[]));
            let jsonl_path = dir.path().join("events.jsonl");

            let mut cmd = Command::new(&exe);
            cmd.env("CELLOS_DEPLOYMENT_PROFILE", "portable")
                .env("CELL_OS_USE_NOOP_SINK", "1")
                .env("CELLOS_CELL_BACKEND", "stub")
                .env("CELLOS_TENANT_ID", tenant)
                .env("CELL_OS_JSONL_EVENTS", &jsonl_path)
                .env("CELLOS_AUTHZ_POLICY_PATH", &authz_path)
                .current_dir(env!("CARGO_MANIFEST_DIR"))
                .arg(&spec_path);
            let output = cmd.output().expect("spawn cellos-supervisor");
            assert!(
                !output.status.success(),
                "tenant {tenant} under deny-all policy must fail; got status {:?}",
                output.status
            );

            let file = std::fs::File::open(&jsonl_path).unwrap_or_else(|e| {
                panic!(
                    "expected JSONL events file at {} for tenant {tenant}: {e}; stderr: {}",
                    jsonl_path.display(),
                    String::from_utf8_lossy(&output.stderr)
                )
            });
            let mut content = String::new();
            for line in BufReader::new(file).lines() {
                let line = line.expect("read JSONL line");
                content.push_str(&line);
                content.push('\n');
            }
            // Keep dir alive until after the read (tempfile cleans on drop).
            drop(dir);
            content
        }

        let jsonl_a = run_tenant("tenantA");
        let jsonl_b = run_tenant("tenantB");

        // Sanity: each run produced at least one CloudEvent line.
        assert!(
            !jsonl_a.trim().is_empty(),
            "tenantA produced no JSONL events"
        );
        assert!(
            !jsonl_b.trim().is_empty(),
            "tenantB produced no JSONL events"
        );

        // Each tenant's JSONL must carry the authz.rejected event with
        // its own tenant id surfaced on the event data.
        assert!(
            jsonl_a.contains("dev.cellos.events.cell.authz.v1.rejected"),
            "tenantA JSONL missing authz.rejected event:\n{jsonl_a}"
        );
        assert!(
            jsonl_b.contains("dev.cellos.events.cell.authz.v1.rejected"),
            "tenantB JSONL missing authz.rejected event:\n{jsonl_b}"
        );

        // Tenant A events MUST carry "tenantA" — on the 1.0 subject axis
        // (`data.subject`) and on the mirrored `data.tenantId`.
        assert!(
            jsonl_a.contains("\"subject\":\"tenantA\""),
            "tenantA JSONL must contain data.subject==tenantA; got:\n{jsonl_a}"
        );
        assert!(
            jsonl_a.contains("\"tenantId\":\"tenantA\""),
            "tenantA JSONL must contain data.tenantId==tenantA; got:\n{jsonl_a}"
        );

        // Tenant B events MUST NOT leak "tenantA" anywhere — proves
        // per-tenant subject namespacing keeps the two streams disjoint.
        assert!(
            !jsonl_b.contains("tenantA"),
            "tenantB JSONL must not contain `tenantA` (cross-tenant leak); got:\n{jsonl_b}"
        );
        // And tenantB must carry its own identifier on the same axes.
        assert!(
            jsonl_b.contains("\"subject\":\"tenantB\""),
            "tenantB JSONL must contain data.subject==tenantB; got:\n{jsonl_b}"
        );
        assert!(
            jsonl_b.contains("\"tenantId\":\"tenantB\""),
            "tenantB JSONL must contain data.tenantId==tenantB; got:\n{jsonl_b}"
        );
    }
}