boxmaker/docs/plans/M3a/files/crates/brokerd/tests/serve.rs

//! `brokerd serve` as a process: its startup, both sockets, and the expiry thread. Do not edit.

#[path = "support/tmp.rs"]
mod tmp;

use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::time::{Duration, Instant};

use brokerd::audit::RECOVERED_NOTICE;
use brokerd::runner::REFUSING;
use proto::{
    AuditEvent, CallId, DenyReason, Empty, Envelope, ErrorCode, Message, PROTOCOL_VERSION,
    ResultStatus, SessionId, ToolRequest, ToolResponse,
};
use tmp::TempDir;

struct Home {
    dir: TempDir,
    config: PathBuf,
}

impl Home {
    fn new(tag: &str, ttl_ms: u64) -> Home {
        let dir = TempDir::new(tag);
        std::fs::create_dir_all(dir.path().join("grants")).unwrap();
        let text = format!(
            "[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n[approvals]\nttl_ms = {ttl_ms}\n",
            home = dir.path().display()
        );
        let config = dir.write("brokerd.toml", &text);
        Home { dir, config }
    }

    fn path(&self, relative: &str) -> PathBuf {
        self.dir.path().join(relative)
    }

    fn tools(&self) -> PathBuf {
        self.path("run/loop-broker/broker.sock")
    }

    fn admin(&self) -> PathBuf {
        self.path("run/owner-broker/admin.sock")
    }

    fn grant(&self, id: &str, mode: &str) {
        let text = format!(
            "tool = \"read_file\"\nmode = \"{mode}\"\nmax_taint = \"secret\"\n\
             result_class = \"private\"\nuntrusted = false\n[constraints]\npaths = [\"/n\"]\n"
        );
        std::fs::write(self.path(&format!("grants/{id}.toml")), text).unwrap();
    }

    fn command(&self, extra: &[&str]) -> Command {
        let mut command = Command::new(env!("CARGO_BIN_EXE_brokerd"));
        command
            .args(["serve", "--config"])
            .arg(&self.config)
            .args(extra);
        command
    }

    /// Starts `brokerd serve` and waits until both sockets answer.
    fn serve(&self) -> Running {
        let child = self
            .command(&[])
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        let running = Running(Some(child));
        let until = Instant::now() + Duration::from_secs(10);
        while UnixStream::connect(self.tools()).is_err()
            || UnixStream::connect(self.admin()).is_err()
        {
            assert!(Instant::now() < until, "brokerd never listened");
            std::thread::sleep(Duration::from_millis(20));
        }
        running
    }

    /// Runs `brokerd serve` expecting it to exit by itself.
    fn run(&self, extra: &[&str]) -> Output {
        self.command(extra).output().unwrap()
    }

    fn events(&self) -> Vec<AuditEvent> {
        let dir = self.path("audit");
        let mut names: Vec<String> = std::fs::read_dir(&dir)
            .unwrap()
            .map(|e| e.unwrap().file_name().into_string().unwrap())
            .filter(|n| n.ends_with(".jsonl"))
            .collect();
        names.sort();
        names
            .iter()
            .flat_map(|n| {
                let text = std::fs::read_to_string(dir.join(n)).unwrap();
                text.lines()
                    .map(|l| serde_json::from_str::<proto::AuditRecord>(l).unwrap().event)
                    .collect::<Vec<_>>()
            })
            .collect()
    }
}

/// Kills the daemon when dropped; `stop` returns what it printed.
struct Running(Option<Child>);

impl Running {
    fn stop(mut self) -> String {
        let mut child = self.0.take().unwrap();
        child.kill().unwrap();
        let output = child.wait_with_output().unwrap();
        String::from_utf8_lossy(&output.stderr).to_string()
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if let Some(child) = &mut self.0 {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

fn mode(path: &Path) -> u32 {
    std::fs::metadata(path).unwrap().permissions().mode() & 0o777
}

fn exchange(socket: &Path, id: u64, msg: Message) -> Vec<Envelope> {
    let mut stream = UnixStream::connect(socket).unwrap();
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .unwrap();
    let env = Envelope {
        v: PROTOCOL_VERSION,
        id,
        r#final: true,
        msg,
    };
    proto::write_frame(&mut stream, &env).unwrap();
    let mut frames = Vec::new();
    loop {
        let env = proto::read_frame(&mut stream).unwrap();
        let last = env.r#final;
        frames.push(env);
        if last {
            return frames;
        }
    }
}

fn read_notes(call: u64) -> Message {
    Message::ToolRequest(ToolRequest {
        session: SessionId::new("s1").unwrap(),
        call: CallId(call),
        tool: "read_file".to_string(),
        arguments: r#"{"path":"/n/a"}"#.to_string(),
    })
}

fn last_response(frames: &[Envelope]) -> &ToolResponse {
    match &frames.last().unwrap().msg {
        Message::ToolResponse(r) => r,
        other => panic!("{other:?}"),
    }
}

fn stderr(output: &Output) -> String {
    String::from_utf8_lossy(&output.stderr).to_string()
}

#[test]
fn it_makes_its_directories_0700_and_its_sockets_0600() {
    let home = Home::new("serve-modes", 900_000);
    // One directory found too open, one made.
    std::fs::create_dir_all(home.path("run/owner-broker")).unwrap();
    std::fs::set_permissions(
        home.path("run/owner-broker"),
        std::fs::Permissions::from_mode(0o755),
    )
    .unwrap();
    let running = home.serve();
    assert_eq!(mode(&home.path("run/loop-broker")), 0o700);
    assert_eq!(mode(&home.path("run/owner-broker")), 0o700);
    assert_eq!(mode(&home.tools()), 0o600);
    assert_eq!(mode(&home.admin()), 0o600);
    assert_eq!(mode(&home.path("audit")), 0o700);
    let printed = running.stop();
    assert!(printed.contains("serving"), "{printed}");
}

#[test]
fn a_stale_socket_is_replaced() {
    let home = Home::new("serve-stale", 900_000);
    std::fs::create_dir_all(home.path("run/loop-broker")).unwrap();
    drop(UnixListener::bind(home.tools()).unwrap());
    assert!(home.tools().exists(), "the stale socket file is there");
    let _running = home.serve();
    let frames = exchange(&home.tools(), 3, read_notes(3));
    assert_eq!(
        last_response(&frames),
        &ToolResponse::Denied {
            reason: DenyReason::NoGrant
        }
    );
}

#[test]
fn a_second_brokerd_on_the_same_home_refuses_to_start() {
    let home = Home::new("serve-twice", 900_000);
    let _running = home.serve();
    let second = home.run(&[]);
    assert_eq!(second.status.code(), Some(1));
    let text = stderr(&second);
    assert!(text.contains("brokerd is already running"), "{text}");
    assert!(
        text.trim_end()
            .ends_with("see docs/runbook.md#brokerd-already-running"),
        "{text}"
    );
    // The first still has its sockets.
    let frames = exchange(&home.admin(), 1, Message::Approvals(Empty {}));
    assert!(matches!(frames[0].msg, Message::ApprovalList(_)));
}

#[test]
fn it_answers_each_socket_and_refuses_the_other_kinds() {
    let home = Home::new("serve-kinds", 900_000);
    home.grant("notes", "auto");
    let running = home.serve();
    let frames = exchange(&home.tools(), 7, read_notes(7));
    assert_eq!(frames[0].id, 7);
    // The production runtime runs nothing.
    assert_eq!(
        last_response(&frames),
        &ToolResponse::Failed {
            message: REFUSING.to_string()
        }
    );
    let wrong = exchange(&home.tools(), 8, Message::Approvals(Empty {}));
    assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden));
    let wrong = exchange(&home.admin(), 9, read_notes(9));
    assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden));
    let printed = running.stop();
    assert!(
        printed.contains("approvals on broker.sock\nsee docs/runbook.md#socket-forbidden"),
        "{printed}"
    );
    assert!(
        printed.contains("tool_request on admin.sock\nsee docs/runbook.md#socket-forbidden"),
        "{printed}"
    );
    assert!(matches!(
        home.events().as_slice(),
        [
            AuditEvent::Decision { .. },
            AuditEvent::Result {
                status: ResultStatus::Failed,
                ..
            }
        ]
    ));
}

#[test]
fn an_approval_nobody_answers_expires() {
    let home = Home::new("serve-expire", 100);
    home.grant("notes", "ask");
    let _running = home.serve();
    let started = Instant::now();
    let frames = exchange(&home.tools(), 2, read_notes(2));
    assert_eq!(frames.len(), 2, "{frames:?}");
    assert!(matches!(
        &frames[0].msg,
        Message::ToolResponse(ToolResponse::PendingApproval { approval: 0, .. })
    ));
    assert_eq!(
        last_response(&frames),
        &ToolResponse::Denied {
            reason: DenyReason::ApprovalExpired
        }
    );
    // The expiry thread looks every second.
    assert!(
        started.elapsed() < Duration::from_secs(5),
        "{:?}",
        started.elapsed()
    );
}

fn copy_case(home: &Home, case: &str, only: &[&str]) {
    let from = format!(
        "{}/../proto/tests/fixtures/audit/{case}",
        env!("CARGO_MANIFEST_DIR")
    );
    std::fs::create_dir_all(home.path("audit")).unwrap();
    for name in only {
        std::fs::copy(
            format!("{from}/{name}"),
            home.path(&format!("audit/{name}")),
        )
        .unwrap();
    }
}

fn snapshot(dir: &Path) -> Vec<(String, Vec<u8>)> {
    let mut all: Vec<(String, Vec<u8>)> = std::fs::read_dir(dir)
        .unwrap()
        .map(|e| e.unwrap())
        .filter(|e| e.file_name().to_string_lossy().ends_with(".jsonl"))
        .map(|e| {
            (
                e.file_name().into_string().unwrap(),
                std::fs::read(e.path()).unwrap(),
            )
        })
        .collect();
    all.sort();
    all
}

#[test]
fn a_broken_chain_stops_it_before_any_socket_and_nothing_is_written() {
    let home = Home::new("serve-broken", 900_000);
    copy_case(&home, "changed-byte", &["2026-09-17.jsonl"]);
    let before = snapshot(&home.path("audit"));
    let output = home.run(&[]);
    assert_eq!(output.status.code(), Some(1));
    let text = stderr(&output);
    assert!(text.contains("2026-09-17.jsonl:4: "), "{text}");
    assert!(
        text.trim_end()
            .ends_with("see docs/runbook.md#audit-chain-broken"),
        "{text}"
    );
    assert_eq!(snapshot(&home.path("audit")), before);
    assert!(!home.tools().exists() && !home.admin().exists());
}

#[test]
fn a_torn_tail_is_recovered_and_it_serves() {
    let home = Home::new("serve-torn", 900_000);
    copy_case(
        &home,
        "torn-tail",
        &["2026-09-17.jsonl", "2026-09-18.jsonl"],
    );
    let running = home.serve();
    let printed = running.stop();
    assert!(printed.contains(RECOVERED_NOTICE), "{printed}");
    assert!(RECOVERED_NOTICE.ends_with("see docs/runbook.md#audit-recovered"));
}

#[test]
fn accept_break_with_nothing_to_accept_exits_2() {
    let home = Home::new("serve-nothing", 900_000);
    let output = home.run(&["--accept-break"]);
    assert_eq!(output.status.code(), Some(2));
    assert!(
        stderr(&output).contains("nothing to accept"),
        "{}",
        stderr(&output)
    );
}

#[test]
fn bad_arguments_and_bad_configs_do_not_start() {
    let brokerd = env!("CARGO_BIN_EXE_brokerd");
    for args in [
        &[][..],
        &["serve"][..],
        &["serve", "--config"][..],
        &["serve", "--config", "a", "--config", "b"][..],
        &["serve", "--config", "a", "--loud"][..],
        &["run", "--config", "a"][..],
    ] {
        let output = Command::new(brokerd).args(args).output().unwrap();
        assert_eq!(output.status.code(), Some(2), "{args:?}");
        assert!(
            stderr(&output).starts_with("usage: brokerd serve"),
            "{args:?}"
        );
    }
    let home = Home::new("serve-config", 900_000);
    std::fs::write(&home.config, "[paths]\nhoem = \"/x\"\n").unwrap();
    let output = home.run(&[]);
    assert_eq!(output.status.code(), Some(1));
    assert!(
        stderr(&output).contains("brokerd.toml"),
        "{}",
        stderr(&output)
    );
    std::fs::remove_file(&home.config).unwrap();
    assert_eq!(home.run(&[]).status.code(), Some(1));
    assert!(
        !home.path("audit").exists(),
        "nothing made before the config is read"
    );
}