From ba8267c7dc0310e47bebbc5090c16c6dfb41f177 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Mon, 28 Sep 2026 01:52:06 -0700 Subject: [PATCH] M5c-1 plan: task 05, brokerd answers memoryd's session-state question on state.sock brokerd serves a third socket, [sockets] state (default /run/memory-broker/state.sock), bound by the same listen (dir 0700, stale socket removed, socket 0600), one thread per connection, one question per connection. The answer is StateStore::read: the stored taint and flag, the store's default (private, trusted) for a session with no file. A damaged state is an internal error with the store's text, never an answer; other kinds are forbidden; malformed frames get read_request's error. A new tests/serve.rs waits for an answer before reading the startup notices (a race that failed about one run in ten under load, before this task too). Checks: every before block found once at ref 04 (20 edits); the new file in the task is byte-identical to the reference; the given tests fail before as the task says (config and state_socket do not compile; serve_state 5 of 5 fail); gate ok with 1171 tests on the reference and on a replay from the task text in a fresh ref-04 worktree with its own build dir (byte-identical to the reference; removed); /tmp unchanged around the gate; 25 mutations, 24 caught, the one missed (state bound last) changes no tested behaviour; the socket tests 25 runs under load, 0 failures; no task-text line over 100 characters outside tables and code. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5c1/05-brokerd-state-socket.md | 627 ++++++++++++++++++ .../M5c1/files/crates/brokerd/tests/config.rs | 138 ++++ .../brokerd/tests/fixtures/config/full.toml | 12 + .../M5c1/files/crates/brokerd/tests/serve.rs | 400 +++++++++++ .../files/crates/brokerd/tests/serve_state.rs | 269 ++++++++ .../crates/brokerd/tests/state_socket.rs | 371 +++++++++++ 6 files changed, 1817 insertions(+) create mode 100644 docs/plans/M5c1/05-brokerd-state-socket.md create mode 100644 docs/plans/M5c1/files/crates/brokerd/tests/config.rs create mode 100644 docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml create mode 100644 docs/plans/M5c1/files/crates/brokerd/tests/serve.rs create mode 100644 docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs create mode 100644 docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs diff --git a/docs/plans/M5c1/05-brokerd-state-socket.md b/docs/plans/M5c1/05-brokerd-state-socket.md new file mode 100644 index 0000000..7ecc61f --- /dev/null +++ b/docs/plans/M5c1/05-brokerd-state-socket.md @@ -0,0 +1,627 @@ +# M5c-1 task 05: `brokerd` answers `memoryd`'s question about a session's state on `state.sock` + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `brokerd: state.sock answers memoryd's question about a session's state` +**Log row:** the Task cell is `M5c1/05-brokerd-state-socket`. + +## Goal + +When a turn ends, `memoryd` will decide where the turn goes in Honcho, and for that it needs the +session's state as `brokerd` knows it: the session's class (its taint) and whether it has read +untrusted content (M5c spec, sections 5 and 7). Task 01 added the two messages: +`bxcore::SessionState { session }`, the question, and `bxcore::SessionStateReply { class, +untrusted }`, the answer. This task makes `brokerd` answer the question on a new socket. +`memoryd` connects to it in task 09. + +- **The socket.** `brokerd` serves a third socket, beside `broker.sock` (tools, from `loopd`) and + `admin.sock` (approvals, from `bxctl`): `state.sock`, by default + `/run/memory-broker/state.sock`. It is configured exactly as the other two are: a key in + the `[sockets]` table of `brokerd.toml`, `state = ""`, where an empty or missing value + means the default under `home`. It is bound by the same `listen` function as the other two, so + its directory is made (and kept) at mode 0700, a stale socket file is removed before the bind, + and the socket is 0600. +- **One question per connection**, as on the other sockets: `memoryd` connects, sends one frame, + reads one final frame with the same `id`, and the connection is closed. A second frame on the + same connection is never read. +- **The answer** comes from the state store, `StateStore::read`, the same files the ledger writes + (`/broker/sessions/.json`): `class` is the stored taint and `untrusted` the stored + flag. A session with no file has made no tool call, and the store answers its default, + `policy::SessionState::default()`, which is `private` and trusted (`untrusted: false`): exactly + what the spec says such a session answers. Asking changes nothing: no file, no directory, no + audit record. +- **Every error is an error frame, never an answer**, so `memoryd` leaves the turn undecided: + + | What arrives, or happens | The final frame | Logged | + |---|---|---| + | `session_state`, the state file reads | `session_state_reply { class, untrusted }` | nothing | + | `session_state`, the file exists but cannot be read (damaged, a directory, …) | `error`, code `internal`, the store's own text (it names the file and ends with `see docs/runbook.md#broker-state-damaged`) | `brokerd: ` and the same text | + | any other kind (a tool request, an approval, a `session_state_reply`, …) | `error`, code `forbidden`, ` is not accepted on state.sock` | `brokerd: refused the message kind on state.sock` and the `socket-forbidden` pointer | + | a malformed frame: not JSON, a session id that is not one, an unknown key, another version | `error`, `bad_message` or `bad_version` (or `bad_frame`), with id 0 | nothing | + + The last two rows are what `broker.sock` and `admin.sock` already do, through the same + functions (`forbid`, `read_request`); nothing new is written for them. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/brokerd/src/config.rs`, `Sockets` | new field `state`, and a doc comment (edit C1) | +| `crates/brokerd/src/config.rs`, `impl Config` | new method `state_socket`, after `admin_socket` (edit C2) | +| `crates/brokerd/src/lib.rs` | `pub mod state_socket;` (edit L1) | +| `crates/brokerd/src/state_socket.rs` | new file: `SOCKET` and `handle` (step 4 gives it whole) | +| `crates/brokerd/src/serve.rs`, `Started`, `start`, `Which`, `run`, `accept_loop`, `serve_one` | the third socket bound, accepted and handled (edits V1 to V11) | +| `crates/brokerd/src/main.rs` | one more line in the serving notice, and the module comment (edits M1 to M3) | +| `docs/runbook.md`, `brokerd-listener-lost` and `socket-forbidden` | they name the third socket (edits R1 to R3) | +| `crates/brokerd/src/broker.rs`, `admin.rs`, `state.rs`, `ledger/` | **no change**: `forbid`, `read_request`, `send` and `StateStore::read` are used as they are | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - new: `crates/brokerd/tests/state_socket.rs`, `crates/brokerd/tests/serve_state.rs` + - **new versions** of (they replace the files in the tree): `crates/brokerd/tests/config.rs` + and `crates/brokerd/tests/fixtures/config/full.toml` (both now set and check `state`), and + `crates/brokerd/tests/serve.rs` (it waits for an answer before it reads the notices; below). +- Create: `crates/brokerd/src/state_socket.rs` (step 4). +- Modify: `crates/brokerd/src/config.rs`, `crates/brokerd/src/lib.rs`, + `crates/brokerd/src/serve.rs`, `crates/brokerd/src/main.rs`, `docs/runbook.md`, + `docs/implementer-log.md`. +- Do not touch: any other source file (`broker.rs`, `admin.rs`, `state.rs`, `ledger/` and + `policy/` in particular), any file of another crate, any `Cargo.toml`, + `crates/brokerd/tests/support/`, any other test. + +## Facts you need + +- **Two types named `SessionState`.** `brokerd::policy::SessionState { taint, untrusted }` is what + the state store returns; `bxcore::SessionState { session }` is the question on the wire. The new + file names **neither** in a `use` line: it matches the question as + `Message::SessionState(question)`, which needs no import, and reads the store's answer through + its fields (`state.taint`, `state.untrusted`), which needs none either. It imports + `bxcore::SessionStateReply` only. Do not add `SessionState` to any `use` line in `brokerd`'s + sources. (The given `state_socket.rs` test needs both, and imports the question as + `bxcore::SessionState as StateQuestion`.) +- **The store's default.** `StateStore::read` returns `Ok(SessionState::default())` when the file + is missing, and `policy::SessionState::default()` is `taint: DataClass::Private, untrusted: + false`. A file that exists and cannot be read, or holds something that is not a state (bad + JSON, an unknown key, a `public` taint), is `Err(StateError)`. `read` never writes: the handler + calls nothing else on the store. +- **Why a new `StateStore` and not the ledger.** The ledger holds the store behind its lock, for + the steps that read and then raise the state. The question only reads, and the store writes each + file atomically (a rename), so a read sees the old state or the new one, whole. The handler + makes its own `StateStore::new(&broker.cfg().state_dir())`, the same directory the ledger's + store is made with in `serve::start`; it never takes the ledger's lock, and so a question never + waits behind a tool call. +- **Threads.** As for the other two sockets: one accept thread for the socket (named + `accept-state`), and one thread per accepted connection (`serve_one`), which runs `handle` once + and ends. A connection that sends nothing holds only its own thread; a given test checks that + the next question is still answered. +- **The order of the binds.** `start` binds `state.sock` **first**, then `broker.sock` and + `admin.sock`, so whoever waits for the tool and admin sockets finds the third bound too. +- **Why a new `tests/serve.rs`.** Its tests started `brokerd`, waited until the two sockets + accepted a connection, then stopped it and looked for the startup notices. A socket accepts as + soon as it is bound, and `brokerd` prints its notices only after that, so under load the notice + was at times not yet printed (`a_torn_tail_is_recovered_and_it_serves` and + `it_makes_its_directories_0700_and_its_sockets_0600` failed about one run in ten, before this + task too). The new version also waits for the admin socket to answer an `approvals` request, + which a handler thread does only after the notices are printed. Nothing else in it changes. +- **The serving notice.** `main` prints one more line after `brokerd: serving tools on … and + approvals on …`: `brokerd: serving session state on `. The existing line keeps its words: + the runbook quotes it. +- **The error for a damaged state.** `StateError`'s `Display` already names the file and ends with + `see docs/runbook.md#broker-state-damaged`, an entry that exists. The handler logs + `brokerd: {e}`, as the ledger does, and sends `e.to_string()` as the error's `detail`. No new + runbook entry: `memoryd`'s pointer for an unanswered question, `broker-state-unavailable`, comes + with its entry in task 09. +- **The given tests.** `state_socket.rs` calls `state_socket::handle` over a socket pair (the + support file `client.rs`, as `broker.rs` does); every read has a 10-second limit, and each test's + home is a `TempDir` removed on drop. `serve_state.rs` starts `brokerd serve` as a process: every + wait has a 10-second deadline, and a guard kills and waits for the process before the home is + removed. On Linux, a Unix socket closed with a frame still unread in it resets the connection, + so the test for "one question per connection" accepts a reset as well as a clean close. +- The calls the edits use: + + | Call | Signature, and what it gives | + |---|---| + | `read_request` | `pub fn read_request(stream: &mut UnixStream) -> Option` (`broker.rs`): the frame, or `None` after answering a malformed one | + | `forbid` | `pub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message, socket: &str)` (`broker.rs`): logs and sends the `forbidden` error | + | `send` | `pub fn send(stream: &mut UnixStream, id: u64, last: bool, msg: Message) -> bool` (`broker.rs`) | + | `StateStore::new` | `pub fn new(dir: &Path) -> StateStore` (`state.rs`) | + | `StateStore::read` | `pub fn read(&self, session: &SessionId) -> Result` (`state.rs`) | + | `Broker::cfg`, `Broker::log` | `pub fn cfg(&self) -> &Config`; `pub fn log(&self, line: &str)` | + | `Config::state_dir` | `pub fn state_dir(&self) -> PathBuf`: `/broker/sessions` | + | `Config::state_socket` | `pub fn state_socket(&self) -> PathBuf` (new, C2) | + | `state_socket::handle` | `pub fn handle(mut stream: UnixStream, broker: &Broker)` (new, step 4) | + | `listen` | `fn listen(socket: &Path) -> Result` (`serve.rs`, unchanged) | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/brokerd/src/config.rs` + +**C1.** The lines + +```rust +/// An empty path means "the default under `home`". +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Default)] +#[serde(deny_unknown_fields, default)] +pub struct Sockets { + pub broker: PathBuf, + pub admin: PathBuf, +} +``` + +become + +```rust +/// The sockets `brokerd` serves. An empty path means "the default under `home`". +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Default)] +#[serde(deny_unknown_fields, default)] +pub struct Sockets { + pub broker: PathBuf, + pub admin: PathBuf, + /// `memoryd`'s session-state questions (M5c spec, section 7). + pub state: PathBuf, +} +``` + +**C2.** The line + +```rust + /// `gateway.sock`, when `[gateway]` is set. +``` + +becomes + +```rust + /// `state.sock`, where `memoryd` asks for a session's state. + pub fn state_socket(&self) -> PathBuf { + if self.sockets.state.as_os_str().is_empty() { + self.paths.home.join("run/memory-broker/state.sock") + } else { + self.sockets.state.clone() + } + } + /// `gateway.sock`, when `[gateway]` is set. +``` + +### `crates/brokerd/src/lib.rs` + +**L1.** The line + +```rust +pub mod state; +``` + +becomes + +```rust +pub mod state; +pub mod state_socket; +``` + +### `crates/brokerd/src/serve.rs` + +**V1.** The line + +```rust +//! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running. +``` + +becomes + +```rust +//! `brokerd serve`: bring the broker up, listen on its three sockets, and keep it running. +``` + +**V2.** The line + +```rust +use crate::state::StateStore; +``` + +becomes + +```rust +use crate::state::StateStore; +use crate::state_socket; +``` + +**V3.** The lines + +```rust +/// The result of `start`: a running broker with both sockets open. +pub struct Started { + pub broker: Arc, + pub recovered: bool, + pub accepted: Option, + tools: UnixListener, + admin: UnixListener, +} + +/// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker. +``` + +become + +```rust +/// The result of `start`: a running broker with its three sockets open. +pub struct Started { + pub broker: Arc, + pub recovered: bool, + pub accepted: Option, + tools: UnixListener, + admin: UnixListener, + state: UnixListener, +} + +/// Start the broker: open the audit chain, bind the three sockets, and build the ledger and broker. +``` + +**V4.** In `start`, the lines + +```rust + // Both sockets, in the order the task gives. + let tools = listen(&cfg.broker_socket())?; +``` + +become + +```rust + // The three sockets. `state.sock` comes first: whoever waits for the tool and admin sockets + // then finds it bound too. + let state = listen(&cfg.state_socket())?; + let tools = listen(&cfg.broker_socket())?; +``` + +**V5.** At the end of `start`, the lines + +```rust + tools, + admin, + }) +} +``` + +become + +```rust + tools, + admin, + state, + }) +} +``` + +**V6.** The lines + +```rust +enum Which { + Tools, + Admin, +} +``` + +become + +```rust +enum Which { + Tools, + Admin, + State, +} +``` + +**V7.** In `run`, the lines + +```rust + tools, + admin, + .. + } = self; +``` + +become + +```rust + tools, + admin, + state, + .. + } = self; +``` + +**V8.** In `run`, the line + +```rust + accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?; +``` + +becomes + +```rust + accept_loop(admin, Arc::clone(&broker), tx.clone(), Which::Admin)?; + accept_loop(state, Arc::clone(&broker), tx, Which::State)?; +``` + +**V9.** Above `accept_loop`, the line + +```rust +/// socket with it. Any other error stops the daemon through `tx`. +``` + +becomes + +```rust +/// sockets with it. Any other error stops the daemon through `tx`. +``` + +**V10.** In `accept_loop`, the lines + +```rust + Which::Admin => "accept-admin", + }; +``` + +become + +```rust + Which::Admin => "accept-admin", + Which::State => "accept-state", + }; +``` + +**V11.** In `serve_one`, the lines + +```rust + Which::Admin => admin::handle(stream, &for_thread), + }); +``` + +become + +```rust + Which::Admin => admin::handle(stream, &for_thread), + Which::State => state_socket::handle(stream, &for_thread), + }); +``` + +### `crates/brokerd/src/main.rs` + +**M1.** The line + +```rust +//! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves. +``` + +becomes + +```rust +//! `brokerd`: the broker role. `serve` takes the audit lock, binds its three sockets, and serves. +``` + +**M2.** The line + +```rust + let admin_path = cfg.admin_socket(); +``` + +becomes + +```rust + let admin_path = cfg.admin_socket(); + let state_path = cfg.state_socket(); +``` + +**M3.** The line + +```rust + eprintln!("brokerd: {runtime_notice}"); +``` + +becomes + +```rust + eprintln!("brokerd: serving session state on {}", state_path.display()); + eprintln!("brokerd: {runtime_notice}"); +``` + +### `docs/runbook.md` + +**R1.** In `## brokerd-listener-lost`, the lines + +```text +- `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: `. It keeps + running and tries every 200 ms; the line is printed once per episode. +``` + +become + +```text +- `brokerd: cannot accept on accept-broker (or accept-admin, accept-state) for now, retrying: + `. It keeps running and tries every 200 ms; the line is printed once per episode. +``` + +**R2.** In `## socket-forbidden`, the line + +```text +requests from `bxctl`. Nothing in the harness sends a wrong kind, so this is a bug or a component +``` + +becomes + +```text +requests from `bxctl`; `state.sock` accepts only `memoryd`'s questions about a session's state. +Nothing in the harness sends a wrong kind, so this is a bug or a component +``` + +**R3.** In `## socket-forbidden`, the lines + +```text +**Confirm.** Note the time, kind and socket. Check what was running: `pgrep -a 'loopd|bxctl'`. +Look at the session logs active at that time. +``` + +become + +```text +**Confirm.** Note the time, kind and socket. Check what was running: +`pgrep -a 'loopd|bxctl|memoryd'`. Look at the session logs active at that time. +``` + +## Steps + +- [ ] **1. Copy the given files.** From the repository root, five commands: + + ```sh + cp docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs crates/brokerd/tests/ + cp docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs crates/brokerd/tests/ + cp docs/plans/M5c1/files/crates/brokerd/tests/config.rs crates/brokerd/tests/ + cp docs/plans/M5c1/files/crates/brokerd/tests/serve.rs crates/brokerd/tests/ + cp docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml \ + crates/brokerd/tests/fixtures/config/ + ``` + + Then `git status --short` lists 3 modified files and 2 new ones, all under + `crates/brokerd/tests/`. +- [ ] **2. See them fail.** Two commands, each with a shell timeout of 600000 ms: + - `cargo build --tests --keep-going -p brokerd`: two test targets do not compile. `config`, + with four `E0599` (``no method named `state_socket` found for struct `Config` ``), and + `state_socket`, with one `E0432` (``unresolved import `brokerd::state_socket` ``). The + new `serve.rs` compiles. + - `cargo test -p brokerd --test serve_state`: it compiles, and its 5 tests fail, after about + 10 s (they run side by side). Four fail with + `brokerd never listened on …/run/memory-broker/state.sock`; + `sockets_state_in_the_config_moves_it` fails with `brokerd never listened on + …/run/loop-broker/broker.sock`, because `brokerd` refuses a config with the unknown key + `state` and exits. +- [ ] **3. The configuration.** Make edits C1 and C2. `cargo test -p brokerd --test config`: + 7 passed. +- [ ] **4. The handler.** Make edit L1, then create `crates/brokerd/src/state_socket.rs` with + exactly this content: + + ```rust + //! One `state.sock` connection: `memoryd` asks for a session's state when a turn ends, and the + //! answer is what the state store holds (M5c spec, section 7). It reads, and never changes + //! anything: no state, no audit record. + + use std::os::unix::net::UnixStream; + + use bxcore::{ErrorCode, Message, SessionStateReply, WireError}; + + use crate::broker::{Broker, forbid, read_request, send}; + use crate::state::StateStore; + + /// The socket's name in the lines `brokerd` logs and the errors it sends. + pub const SOCKET: &str = "state.sock"; + + /// Serve one `state.sock` connection from `memoryd`: one question, one final answer, then the + /// connection is closed. + pub fn handle(mut stream: UnixStream, broker: &Broker) { + // 1. The peer is gone before it speaks, or its frame is malformed: answered, then return. + let Some(envelope) = read_request(&mut stream) else { + return; + }; + let id = envelope.id; + let msg = envelope.msg; + + // 2. Only the question belongs here. `bxcore::SessionState` is matched through `Message`, so + // this file names neither it nor `policy::SessionState`, which the store returns. + let Message::SessionState(question) = msg else { + forbid(broker, &mut stream, id, &msg, SOCKET); + return; + }; + + // 3. The store the ledger writes. A session without a file is the store's default, `private` + // and trusted; a file that cannot be read is an error, never that default. + let store = StateStore::new(&broker.cfg().state_dir()); + let response = match store.read(&question.session) { + Ok(state) => Message::SessionStateReply(SessionStateReply { + class: state.taint, + untrusted: state.untrusted, + }), + Err(e) => { + broker.log(&format!("brokerd: {e}")); + Message::Error(WireError { + code: ErrorCode::Internal, + detail: e.to_string(), + }) + } + }; + + // 4. The final frame; a failed send is ignored, as the peer is gone. + let _ = send(&mut stream, id, true, response); + } + ``` + + `cargo check -p brokerd`: no error and no warning. Then + `cargo test -p brokerd --test state_socket`: 8 passed. +- [ ] **5. Serve it.** Make edits V1 to V11 in `serve.rs`, then M1 to M3 in `main.rs`. + `cargo check -p brokerd`: no error and no warning. Then + `cargo test -p brokerd --test serve_state`: 5 passed. Then + `cargo test -p brokerd --test serve --test instance --test serve_pointers`: every test passes + (8, 2 and 10). + The new tests: + + | Test | What it checks | + |---|---| + | `the_answer_is_the_stored_class_and_flag` | step 4: class and flag, all four pairs, from states written by the store's own `raise` | + | `a_state_file_the_owner_wrote_is_answered_as_it_says` | step 4: a file written by hand | + | `a_session_brokerd_has_not_seen_is_private_and_trusted` | step 4: the store's default; no file or directory made | + | `asking_changes_nothing` | step 4: state files, audit log and log lines unchanged | + | `a_damaged_state_is_an_error_never_an_answer` | step 4: `internal`, the store's text, logged once | + | `every_other_kind_is_forbidden_and_logged` | step 4: `forbid` with `SOCKET` | + | `a_malformed_frame_is_answered_with_an_error` | step 4: `read_request`'s answers, a path as a session id | + | `a_second_question_on_one_connection_is_not_answered` | step 4: one question per connection | + | `it_answers_on_its_default_path_from_the_state_files` | C2, V3 to V11, M2, M3, end to end | + | `its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replaced` | V4: bound by `listen` | + | `sockets_state_in_the_config_moves_it` | C1, C2: `[sockets] state` | + | `a_silent_connection_does_not_hold_up_the_next` | V11: a thread per connection | + | `the_other_sockets_refuse_the_question` | the tool and admin sockets forbid it (no change) | + + The new versions of `config.rs` and `full.toml` check `state_socket` by default, when set, and + beside the other two. +- [ ] **6. The runbook.** Make edits R1, R2 and R3. +- [ ] **7. Check.** Four commands: + - `grep -n 'listen(&cfg' crates/brokerd/src/serve.rs` prints three lines, `state_socket` first. + - `grep -n '^use' crates/brokerd/src/state_socket.rs` prints four lines; the one from `bxcore` + imports `ErrorCode`, `Message`, `SessionStateReply` and `WireError`, and no `SessionState`. + - `git diff --stat -- crates/brokerd/src` lists exactly `config.rs`, `lib.rs`, `main.rs` and + `serve.rs` (the new `state_socket.rs` is untracked until you stage it). + - `git diff --stat -- crates/bxcore crates/loopd crates/memoryd crates/gatewayd crates/bxctl` + prints nothing. +- [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1171 tests: 1158 before this task, plus 13 new ones (eight + in `state_socket.rs`, five in `serve_state.rs`; the new `config.rs` and `serve.rs` keep their + counts, 7 and 8). +- [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/05-brokerd-state-socket`. `git status --short` lists the four sources of "Modify" + under `crates/brokerd/src/`, the three replaced test files, `docs/runbook.md` and your log as + modified (ten files), and `crates/brokerd/src/state_socket.rs` and the two new test files as + new. Stage with + `git add crates/brokerd docs/runbook.md docs/implementer-log.md Cargo.lock`, then commit as + `AGENTS.md` says. `Cargo.lock` does not change in this task; the `git add` is harmless. + +## Done when + +- `make gate` prints `gate: ok` with 1171 tests. +- `git diff HEAD~1 --stat` lists only the four modified sources, `state_socket.rs`, the five + copied files, `docs/runbook.md` and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of any edit is not in its file exactly as shown, or is there more than once. +- Step 2 shows a different result: other test targets, errors of another kind, or other tests + failing in `serve_state`. +- `cargo check` reports an error or a warning after step 4 or 5. +- After step 3, 4 or 5 a test fails. Do not change a test, a fixture or the support files: stop, + and give the failing tests' names and their messages. +- The fix seems to need a change anywhere else: `broker.rs`, `state.rs`, the ledger, another + crate, a default answer when the state cannot be read, a second question on one connection, a + new runbook entry, or `SessionState` in a `use` line. It does not; stop and say why you think + so. +- A test in `inferproxy`, `toolkit`, `loopd` or `memoryd` fails in the gate: check once whether it + fails without your change, then stop and report both results. +- The gate passes with a count other than 1171. +- A comment in an edit or in the new file does not fit the code under it. Quote it and stop. diff --git a/docs/plans/M5c1/files/crates/brokerd/tests/config.rs b/docs/plans/M5c1/files/crates/brokerd/tests/config.rs new file mode 100644 index 0000000..1c44331 --- /dev/null +++ b/docs/plans/M5c1/files/crates/brokerd/tests/config.rs @@ -0,0 +1,138 @@ +//! Tests for `brokerd`'s configuration. Do not edit: these define the required behaviour. + +use brokerd::config::{Approvals, Config, ConfigError, Sockets}; +use std::path::{Path, PathBuf}; + +fn fixture(name: &str) -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/fixtures/config") + .join(name) +} + +/// What `home` must default to in this process. The test does not set the variable: changing the +/// environment of a running test binary would race with the other tests. +fn default_home() -> PathBuf { + std::env::var_os("BOXMAKER_HOME") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/var/lib/boxmaker")) +} + +#[test] +fn an_empty_file_gets_every_default() { + let c = Config::load(&fixture("empty.toml")).unwrap(); + assert_eq!(c.paths.home, default_home()); + assert_eq!(c.paths.grants, PathBuf::from("/etc/boxmaker/grants")); + assert_eq!(c.sockets, Sockets::default()); + assert_eq!(c.approvals, Approvals { ttl_ms: 900_000 }); + assert_eq!(Approvals::default(), Approvals { ttl_ms: 900_000 }); + assert_eq!(c, Config::parse("").unwrap()); + assert_eq!(c, Config::default()); +} + +#[test] +fn sockets_and_directories_default_to_places_under_home() { + let c = Config::load(&fixture("home_only.toml")).unwrap(); + assert_eq!(c.paths.home, PathBuf::from("/srv/boxmaker")); + assert_eq!( + c.broker_socket(), + PathBuf::from("/srv/boxmaker/run/loop-broker/broker.sock") + ); + assert_eq!( + c.admin_socket(), + PathBuf::from("/srv/boxmaker/run/owner-broker/admin.sock") + ); + assert_eq!( + c.state_socket(), + PathBuf::from("/srv/boxmaker/run/memory-broker/state.sock") + ); + assert_eq!(c.audit_dir(), PathBuf::from("/srv/boxmaker/audit")); + assert_eq!( + c.state_dir(), + PathBuf::from("/srv/boxmaker/broker/sessions") + ); + // The grants are not under home: the owner writes them, brokerd only reads them. + assert_eq!(c.paths.grants, PathBuf::from("/etc/boxmaker/grants")); +} + +#[test] +fn every_key_can_be_set() { + let c = Config::load(&fixture("full.toml")).unwrap(); + assert_eq!(c.paths.home, PathBuf::from("/srv/boxmaker")); + assert_eq!(c.paths.grants, PathBuf::from("/srv/boxmaker-grants")); + assert_eq!(c.broker_socket(), PathBuf::from("/run/bx/broker.sock")); + assert_eq!(c.admin_socket(), PathBuf::from("/run/bx/admin.sock")); + assert_eq!(c.state_socket(), PathBuf::from("/run/bx/state.sock")); + assert_eq!(c.approvals.ttl_ms, 60_000); + // The two directories always follow home. + assert_eq!(c.audit_dir(), PathBuf::from("/srv/boxmaker/audit")); + assert_eq!( + c.state_dir(), + PathBuf::from("/srv/boxmaker/broker/sessions") + ); +} + +#[test] +fn one_socket_set_leaves_the_other_at_its_default() { + let c = + Config::parse("[paths]\nhome = \"/h\"\n[sockets]\nadmin = \"/x/admin.sock\"\n").unwrap(); + assert_eq!(c.admin_socket(), PathBuf::from("/x/admin.sock")); + assert_eq!( + c.broker_socket(), + PathBuf::from("/h/run/loop-broker/broker.sock") + ); + assert_eq!( + c.state_socket(), + PathBuf::from("/h/run/memory-broker/state.sock") + ); + let c = + Config::parse("[paths]\nhome = \"/h\"\n[sockets]\nstate = \"/x/state.sock\"\n").unwrap(); + assert_eq!(c.state_socket(), PathBuf::from("/x/state.sock")); + assert_eq!( + c.broker_socket(), + PathBuf::from("/h/run/loop-broker/broker.sock") + ); + assert_eq!( + c.admin_socket(), + PathBuf::from("/h/run/owner-broker/admin.sock") + ); +} + +#[test] +fn unknown_keys_and_tables_are_errors() { + for name in ["unknown_key.toml", "unknown_table.toml", "wrong_type.toml"] { + match Config::load(&fixture(name)) { + Err(ConfigError::Parse(path, _)) => assert_eq!(path, fixture(name)), + other => panic!("{name}: expected a parse error, got {other:?}"), + } + } + // In every table, not only the one the fixture shows. + for text in [ + "[paths]\nhome = \"/h\"\nhouse = \"/h\"\n", + "[sockets]\nbroker = \"/b.sock\"\nloop = \"/l.sock\"\n", + "[approvals]\nttl_ms = 1\nttl_s = 1\n", + "top = 1\n", + "[approvals]\nttl_ms = -5\n", + ] { + assert!(Config::parse(text).is_err(), "accepted: {text}"); + } +} + +#[test] +fn a_missing_file_is_a_read_error_that_names_the_file() { + let path = fixture("does-not-exist.toml"); + match Config::load(&path) { + Err(ConfigError::Read(p, _)) => assert_eq!(p, path), + other => panic!("expected a read error, got {other:?}"), + } + let text = Config::load(&path).unwrap_err().to_string(); + assert!(text.contains("does-not-exist.toml"), "{text}"); +} + +#[test] +fn a_parse_error_names_the_file_and_the_key() { + let text = Config::load(&fixture("unknown_key.toml")) + .unwrap_err() + .to_string(); + assert!(text.contains("unknown_key.toml"), "{text}"); + assert!(text.contains("ttl"), "{text}"); +} diff --git a/docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml b/docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml new file mode 100644 index 0000000..8d9dc39 --- /dev/null +++ b/docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml @@ -0,0 +1,12 @@ +# Every key set. +[paths] +home = "/srv/boxmaker" +grants = "/srv/boxmaker-grants" + +[sockets] +broker = "/run/bx/broker.sock" +admin = "/run/bx/admin.sock" +state = "/run/bx/state.sock" + +[approvals] +ttl_ms = 60000 diff --git a/docs/plans/M5c1/files/crates/brokerd/tests/serve.rs b/docs/plans/M5c1/files/crates/brokerd/tests/serve.rs new file mode 100644 index 0000000..056b657 --- /dev/null +++ b/docs/plans/M5c1/files/crates/brokerd/tests/serve.rs @@ -0,0 +1,400 @@ +//! `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 bxcore::{ + 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. A socket accepts connections + /// as soon as it is bound, before `brokerd` has printed its startup notices; the admin + /// socket's answer comes from a handler thread, which runs only after they are printed. + 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)); + } + let answer = exchange(&self.admin(), 1, Message::Approvals(Empty {})); + assert!( + matches!(&answer[0].msg, Message::ApprovalList(_)), + "{answer:?}" + ); + 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 { + let dir = self.path("audit"); + let mut names: Vec = 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::(l) + .unwrap() + .event + }) + .collect::>() + }) + .collect() + } +} + +/// Kills the daemon when dropped; `stop` returns what it printed. +struct Running(Option); + +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 { + 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, + }; + bxcore::write_frame(&mut stream, &env).unwrap(); + let mut frames = Vec::new(); + loop { + let env = bxcore::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, + detail: None, + } + ); +} + +#[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, + detail: None, + } + ); + // 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!( + "{}/../bxcore/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)> { + let mut all: Vec<(String, Vec)> = 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" + ); +} diff --git a/docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs b/docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs new file mode 100644 index 0000000..40bc104 --- /dev/null +++ b/docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs @@ -0,0 +1,269 @@ +//! `brokerd serve` answers `memoryd`'s session-state question on its third socket, +//! `run/memory-broker/state.sock` or `[sockets] state` (M5c spec, section 7). 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, Stdio}; +use std::time::{Duration, Instant}; + +use bxcore::{ + DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, + SessionState as StateQuestion, SessionStateReply, +}; +use tmp::TempDir; + +struct Home { + dir: TempDir, + config: PathBuf, +} + +impl Home { + /// A home whose config has `sockets` as the body of its `[sockets]` table. + fn new(tag: &str, sockets: &str) -> 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[sockets]\n{sockets}\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 state(&self) -> PathBuf { + self.path("run/memory-broker/state.sock") + } + + /// Writes the session's state file, as the ledger would. + fn session(&self, id: &str, text: &str) { + std::fs::create_dir_all(self.path("broker/sessions")).unwrap(); + std::fs::write(self.path(&format!("broker/sessions/{id}.json")), text).unwrap(); + } + + /// Starts `brokerd serve` and waits until the tool socket, the admin socket and `state` + /// accept a connection. + fn serve(&self, state: &Path) -> Running { + let child = Command::new(env!("CARGO_BIN_EXE_brokerd")) + .args(["serve", "--config"]) + .arg(&self.config) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let running = Running(Some(child)); + let until = Instant::now() + Duration::from_secs(10); + for socket in [ + self.path("run/loop-broker/broker.sock"), + self.path("run/owner-broker/admin.sock"), + state.to_path_buf(), + ] { + while UnixStream::connect(&socket).is_err() { + assert!( + Instant::now() < until, + "brokerd never listened on {}", + socket.display() + ); + std::thread::sleep(Duration::from_millis(20)); + } + } + running + } +} + +/// Kills the daemon when dropped, and waits for it; `stop` returns what it printed. +struct Running(Option); + +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 +} + +/// A connection with a ten-second read limit, so a daemon that never answers fails the test. +fn connect(socket: &Path) -> UnixStream { + let stream = UnixStream::connect(socket).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + stream +} + +/// Sends one message under `id` and returns the one final frame that answers it. +fn exchange(socket: &Path, id: u64, msg: Message) -> Envelope { + let mut stream = connect(socket); + bxcore::write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg, + }, + ) + .unwrap(); + let env = bxcore::read_frame(&mut stream).unwrap(); + assert_eq!((env.id, env.r#final), (id, true), "{env:?}"); + env +} + +fn question(session: &str) -> Message { + Message::SessionState(StateQuestion { + session: SessionId::new(session).unwrap(), + }) +} + +fn reply(env: &Envelope) -> SessionStateReply { + match &env.msg { + Message::SessionStateReply(reply) => reply.clone(), + other => panic!("wanted a session_state_reply, got {other:?}"), + } +} + +#[test] +fn it_answers_on_its_default_path_from_the_state_files() { + let home = Home::new("serve-state-answers", ""); + home.session("s1", "{\"taint\":\"secret\",\"untrusted\":true}\n"); + home.session("s2", "{\"taint\":\"private\",\"untrusted\":true}\n"); + let running = home.serve(&home.state()); + for (id, session, class, untrusted) in [ + (1, "s1", DataClass::Secret, true), + (2, "s2", DataClass::Private, true), + (3, "unseen", DataClass::Private, false), + ] { + assert_eq!( + reply(&exchange(&home.state(), id, question(session))), + SessionStateReply { class, untrusted }, + "{session}" + ); + } + assert!(!home.path("broker/sessions/unseen.json").exists()); + let printed = running.stop(); + assert!( + printed.contains(&format!( + "brokerd: serving session state on {}\n", + home.state().display() + )), + "{printed}" + ); +} + +/// As for the other two sockets: the directory is made private even when it was there at 0755, +/// the socket is 0600, and a stale socket left by a killed `brokerd` is replaced. +#[test] +fn its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replaced() { + let home = Home::new("serve-state-modes", ""); + std::fs::create_dir_all(home.path("run/memory-broker")).unwrap(); + std::fs::set_permissions( + home.path("run/memory-broker"), + std::fs::Permissions::from_mode(0o755), + ) + .unwrap(); + drop(UnixListener::bind(home.state()).unwrap()); + assert!(home.state().exists(), "the stale socket file is there"); + let _running = home.serve(&home.state()); + assert_eq!(mode(&home.path("run/memory-broker")), 0o700); + assert_eq!(mode(&home.state()), 0o600); + assert_eq!( + reply(&exchange(&home.state(), 9, question("s9"))), + SessionStateReply { + class: DataClass::Private, + untrusted: false, + } + ); +} + +#[test] +fn sockets_state_in_the_config_moves_it() { + let dir = TempDir::new("serve-state-moved-sock"); + let moved = dir.path().join("st/state.sock"); + let home = Home::new( + "serve-state-moved", + &format!("state = \"{}\"", moved.display()), + ); + home.session("s1", "{\"taint\":\"secret\",\"untrusted\":false}\n"); + let running = home.serve(&moved); + assert_eq!( + reply(&exchange(&moved, 1, question("s1"))), + SessionStateReply { + class: DataClass::Secret, + untrusted: false, + } + ); + assert_eq!(mode(&dir.path().join("st")), 0o700); + assert!( + !home.path("run/memory-broker").exists(), + "nothing at the default" + ); + let printed = running.stop(); + assert!( + printed.contains(&format!( + "brokerd: serving session state on {}\n", + moved.display() + )), + "{printed}" + ); +} + +/// A connection that sends nothing holds only its own thread: the next question is answered. +#[test] +fn a_silent_connection_does_not_hold_up_the_next() { + let home = Home::new("serve-state-silent", ""); + let _running = home.serve(&home.state()); + let _silent = connect(&home.state()); + assert_eq!( + reply(&exchange(&home.state(), 2, question("s2"))), + SessionStateReply { + class: DataClass::Private, + untrusted: false, + } + ); +} + +/// The question belongs on `state.sock` only: the tool and admin sockets refuse it. +#[test] +fn the_other_sockets_refuse_the_question() { + let home = Home::new("serve-state-others", ""); + let running = home.serve(&home.state()); + for (id, socket) in [ + (4, home.path("run/loop-broker/broker.sock")), + (5, home.path("run/owner-broker/admin.sock")), + ] { + match exchange(&socket, id, question("s1")).msg { + Message::Error(e) => assert_eq!(e.code, ErrorCode::Forbidden, "{e:?}"), + other => panic!("{other:?}"), + } + } + let printed = running.stop(); + assert!( + printed.contains("session_state on broker.sock\nsee docs/runbook.md#socket-forbidden"), + "{printed}" + ); + assert!( + printed.contains("session_state on admin.sock\nsee docs/runbook.md#socket-forbidden"), + "{printed}" + ); +} diff --git a/docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs b/docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs new file mode 100644 index 0000000..c604350 --- /dev/null +++ b/docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs @@ -0,0 +1,371 @@ +//! One `state.sock` connection: `memoryd` asks for a session's state, and `brokerd` answers from +//! its state store, changing nothing (M5c spec, section 7). Every error is an error frame, never +//! an answer. Do not edit. + +#[path = "support/client.rs"] +mod client; +#[path = "support/rig.rs"] +mod rig; +#[path = "support/runtime.rs"] +mod runtime; +#[path = "support/sink.rs"] +mod sink; +#[path = "support/tmp.rs"] +mod tmp; + +use std::io::Write; +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use brokerd::broker::Broker; +use brokerd::policy::{Label, SessionState}; +use brokerd::state_socket; +use bxcore::{ + CallId, DataClass, Empty, Envelope, ErrorCode, FrameError, Message, PROTOCOL_VERSION, + SessionId, SessionState as StateQuestion, SessionStateReply, SessionUpdated, ToolRequest, + WireError, +}; +use client::{Serve, next, open}; +use rig::Rig; +use runtime::Recording; + +fn broker(rig: &Rig) -> Arc { + rig.broker(&Recording::answering("x")) +} + +fn question(session: &str) -> Message { + Message::SessionState(StateQuestion { + session: SessionId::new(session).unwrap(), + }) +} + +/// Asks once, under `id`, and returns the one frame that comes back. +fn ask(broker: &Arc, id: u64, msg: Message) -> Envelope { + let mut stream = open(broker, state_socket::handle, id, msg); + let env = next(&mut stream); + assert_eq!((env.id, env.r#final), (id, true), "{env:?}"); + env +} + +fn reply(env: &Envelope) -> SessionStateReply { + match &env.msg { + Message::SessionStateReply(reply) => reply.clone(), + other => panic!("wanted a session_state_reply, got {other:?}"), + } +} + +fn error(env: &Envelope) -> WireError { + match &env.msg { + Message::Error(e) => e.clone(), + other => panic!("wanted an error, got {other:?}"), + } +} + +/// Raises a session's state the way the ledger does, through the store's own `raise`. +fn raise(rig: &Rig, session: &str, class: DataClass, untrusted: bool) { + rig.state() + .raise( + &SessionId::new(session).unwrap(), + SessionState::default(), + Label { class, untrusted }, + ) + .unwrap(); +} + +/// Every file under `dir`, with its bytes, sorted; empty when `dir` does not exist. +fn files(dir: &Path) -> Vec<(String, Vec)> { + let mut all = Vec::new(); + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries { + let entry = entry.unwrap(); + all.push(( + entry.file_name().into_string().unwrap(), + std::fs::read(entry.path()).unwrap_or_default(), + )); + } + } + all.sort(); + all +} + +#[test] +fn the_answer_is_the_stored_class_and_flag() { + let rig = Rig::new("state-sock-stored"); + raise(&rig, "a", DataClass::Private, false); + raise(&rig, "b", DataClass::Private, true); + raise(&rig, "c", DataClass::Secret, false); + raise(&rig, "d", DataClass::Secret, true); + // A public result leaves a session at private: a session is never below private. + raise(&rig, "e", DataClass::Public, true); + let broker = broker(&rig); + for (i, (session, class, untrusted)) in [ + ("a", DataClass::Private, false), + ("b", DataClass::Private, true), + ("c", DataClass::Secret, false), + ("d", DataClass::Secret, true), + ("e", DataClass::Private, true), + ] + .into_iter() + .enumerate() + { + let id = 40 + i as u64; + assert_eq!( + reply(&ask(&broker, id, question(session))), + SessionStateReply { class, untrusted }, + "session {session}" + ); + } + assert!(rig.lines.all().is_empty(), "{:?}", rig.lines.all()); +} + +/// A file written by hand, as the runbook tells the owner to write one, is read the same way. +#[test] +fn a_state_file_the_owner_wrote_is_answered_as_it_says() { + let rig = Rig::new("state-sock-owner"); + std::fs::create_dir_all(rig.cfg.state_dir()).unwrap(); + std::fs::write( + rig.state_file("s1"), + "{\"taint\":\"secret\",\"untrusted\":true}\n", + ) + .unwrap(); + let env = ask(&broker(&rig), 7, question("s1")); + assert_eq!( + reply(&env), + SessionStateReply { + class: DataClass::Secret, + untrusted: true, + } + ); +} + +/// A session `brokerd` has not seen has made no tool call: `private`, trusted (M5c spec, section +/// 7). It is answered without making its file, or the directory. +#[test] +fn a_session_brokerd_has_not_seen_is_private_and_trusted() { + let rig = Rig::new("state-sock-unseen"); + let broker = broker(&rig); + let env = ask(&broker, 3, question("never-seen")); + assert_eq!( + reply(&env), + SessionStateReply { + class: DataClass::Private, + untrusted: false, + } + ); + assert!(!rig.cfg.state_dir().exists(), "asking makes no directory"); + // With the directory there and another session in it, still no file for this one. + raise(&rig, "other", DataClass::Secret, true); + let env = ask(&broker, 4, question("never-seen")); + assert_eq!( + reply(&env), + SessionStateReply { + class: DataClass::Private, + untrusted: false, + } + ); + assert!(!rig.state_file("never-seen").exists()); +} + +#[test] +fn asking_changes_nothing() { + let rig = Rig::new("state-sock-nothing"); + raise(&rig, "s1", DataClass::Secret, true); + raise(&rig, "s2", DataClass::Private, false); + let before = files(&rig.cfg.state_dir()); + let broker = broker(&rig); + for id in 0..3 { + for session in ["s1", "s2", "s3"] { + ask(&broker, id, question(session)); + } + } + assert_eq!(files(&rig.cfg.state_dir()), before); + assert!( + rig.events().is_empty(), + "nothing is recorded for a question" + ); + assert!(rig.lines.all().is_empty(), "{:?}", rig.lines.all()); +} + +/// A state file that cannot be read is an `internal` error with the store's reason and runbook +/// pointer, logged once, never an answer: `memoryd` then leaves the turn undecided. +#[test] +fn a_damaged_state_is_an_error_never_an_answer() { + let cases: [(&str, Option<&str>); 5] = [ + ("not json", Some("{\"taint\":")), + ( + "public", + Some("{\"taint\":\"public\",\"untrusted\":false}\n"), + ), + ( + "unknown key", + Some("{\"taint\":\"private\",\"untrusted\":false,\"x\":1}\n"), + ), + ( + "unknown class", + Some("{\"taint\":\"loud\",\"untrusted\":false}\n"), + ), + ("a directory", None), + ]; + for (i, (name, text)) in cases.into_iter().enumerate() { + let rig = Rig::new("state-sock-damaged"); + std::fs::create_dir_all(rig.cfg.state_dir()).unwrap(); + match text { + Some(text) => std::fs::write(rig.state_file("s1"), text).unwrap(), + None => std::fs::create_dir(rig.state_file("s1")).unwrap(), + } + let before = files(&rig.cfg.state_dir()); + let env = ask(&broker(&rig), 60 + i as u64, question("s1")); + let e = error(&env); + assert_eq!(e.code, ErrorCode::Internal, "{name}: {e:?}"); + assert!(e.detail.contains("s1.json"), "{name}: {}", e.detail); + assert!( + e.detail + .ends_with("see docs/runbook.md#broker-state-damaged"), + "{name}: {}", + e.detail + ); + let lines = rig.lines.all(); + assert_eq!(lines.len(), 1, "{name}: {lines:?}"); + assert!(lines[0].starts_with("brokerd: "), "{name}: {lines:?}"); + assert!(lines[0].contains("s1.json"), "{name}: {lines:?}"); + assert!( + lines[0].ends_with("see docs/runbook.md#broker-state-damaged"), + "{name}: {lines:?}" + ); + assert_eq!(files(&rig.cfg.state_dir()), before, "{name}"); + } +} + +#[test] +fn every_other_kind_is_forbidden_and_logged() { + let rig = Rig::new("state-sock-forbidden"); + raise(&rig, "s1", DataClass::Secret, true); + let broker = broker(&rig); + let session = SessionId::new("s1").unwrap(); + let kinds = vec![ + Message::ToolRequest(ToolRequest { + session: session.clone(), + call: CallId(1), + tool: "read_file".to_string(), + arguments: "{}".to_string(), + }), + Message::Approvals(Empty {}), + Message::CheckGrants(Empty {}), + Message::Ok(Empty {}), + Message::SessionUpdated(SessionUpdated { + session: session.clone(), + turn: 1, + }), + // An answer sent as a question is not a question. + Message::SessionStateReply(SessionStateReply { + class: DataClass::Private, + untrusted: false, + }), + ]; + let count = kinds.len(); + for (i, msg) in kinds.into_iter().enumerate() { + let id = 100 + i as u64; + let e = error(&ask(&broker, id, msg)); + assert_eq!(e.code, ErrorCode::Forbidden, "{e:?}"); + assert!(e.detail.ends_with("is not accepted on state.sock"), "{e:?}"); + } + let lines = rig.lines.with("on state.sock"); + assert_eq!(lines.len(), count, "{:?}", rig.lines.all()); + for line in &lines { + assert!( + line.ends_with("\nsee docs/runbook.md#socket-forbidden"), + "{line}" + ); + } + assert_eq!( + rig.lines + .with("refused the message kind session_state_reply on state.sock") + .len(), + 1 + ); + assert!(rig.events().is_empty(), "nothing is recorded"); +} + +/// A frame that is not a well-formed question is answered with the frame error, id 0, as on the +/// other sockets. A session id that is not one (a path, say) is refused when the frame is read. +#[test] +fn a_malformed_frame_is_answered_with_an_error() { + let rig = Rig::new("state-sock-malformed"); + raise(&rig, "s1", DataClass::Secret, true); + let broker = broker(&rig); + let body = |session: &str, v: u32, extra: &str| { + format!( + "{{\"v\":{v},\"id\":5,\"final\":true,\"msg\":{{\"kind\":\"session_state\",\ + \"body\":{{\"session\":\"{session}\"{extra}}}}}}}" + ) + }; + let cases = [ + ("not json", "{{{".to_string(), ErrorCode::BadMessage), + ("a path", body("../s1", 1, ""), ErrorCode::BadMessage), + ("upper case", body("S1", 1, ""), ErrorCode::BadMessage), + ( + "unknown key", + body("s1", 1, ",\"x\":1"), + ErrorCode::BadMessage, + ), + ("version 2", body("s1", 2, ""), ErrorCode::BadVersion), + ]; + for (name, text, code) in cases { + let (mut client, server) = UnixStream::pair().unwrap(); + let b = Arc::clone(&broker); + std::thread::spawn(move || state_socket::handle(server, &b)); + client + .write_all(&(text.len() as u32).to_be_bytes()) + .unwrap(); + client.write_all(text.as_bytes()).unwrap(); + let env = next(&mut client); + assert_eq!((env.id, env.r#final), (0, true), "{name}"); + assert_eq!(error(&env).code, code, "{name}"); + } + // The well-formed question, for contrast. + let env = ask(&broker, 5, question("s1")); + assert_eq!( + reply(&env), + SessionStateReply { + class: DataClass::Secret, + untrusted: true, + } + ); +} + +/// One question per connection: after its answer the connection is closed, and a second question +/// on it is never answered. +#[test] +fn a_second_question_on_one_connection_is_not_answered() { + let rig = Rig::new("state-sock-once"); + raise(&rig, "s1", DataClass::Secret, true); + let broker = broker(&rig); + let mut stream = open(&broker, state_socket::handle, 1, question("s1")); + assert!(matches!( + next(&mut stream).msg, + Message::SessionStateReply(_) + )); + // The handler may have closed its end already, so the second frame may not be written. + let _ = bxcore::write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: 2, + r#final: true, + msg: question("s1"), + }, + ); + // Once the peer has closed, macOS refuses the timeout with EINVAL (22); the read cannot block + // then. + if let Err(e) = stream.set_read_timeout(Some(Duration::from_secs(10))) { + assert_eq!(e.raw_os_error(), Some(22), "set_read_timeout: {e}"); + } + match bxcore::read_frame(&mut stream) { + Err(FrameError::Closed) => {} + // Linux resets a Unix socket closed with unread data in it: closed as well. + Err(FrameError::Io(e)) if e.kind() == std::io::ErrorKind::ConnectionReset => {} + other => panic!("wanted the connection closed, got {other:?}"), + } +}