diff --git a/docs/plans/M5b/01-invalid-arguments-detail.md b/docs/plans/M5b/01-invalid-arguments-detail.md new file mode 100644 index 0000000..50fb8b3 --- /dev/null +++ b/docs/plans/M5b/01-invalid-arguments-detail.md @@ -0,0 +1,551 @@ +# M5b task 01: an invalid call is denied with its reason + +**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `bxcore, brokerd, loopd: an invalid-arguments denial says why` + +## Goal + +`brokerd` refuses a tool call whose arguments do not parse (`invalid_arguments`). Its parser +explains why (`ArgsError`), but `policy::decide` throws the explanation away, so the model reads +only a fixed sentence and has to guess. On the device (2026-09-25) it spent two calls finding that +`note` takes no `file`. After this task the explanation reaches the model (M5b spec, section 4): + +| The model's call to `memory` | What it reads back | +|---|---| +| `{"action":"note","file":"notes.md","text":"x"}`, before | `Denied: the arguments are not valid for this tool.` | +| the same, after | ``Denied: the arguments are not valid for this tool: invalid argument shape: unknown field `file`, expected `text`.`` | + +Four parts: + +1. **`bxcore`**: `ToolResponse::Denied` gains a field `detail: Option`, written only when + it is `Some`. A denial without a detail has exactly the bytes it had before, so every existing + fixture stays valid. One new fixture has a detail. +2. **`brokerd`**: `ArgsError` gains `detail()`: its `Display` text on one line (each control + character, and U+2028 and U+2029, becomes a space), cut to 300 characters. `policy::Denial` + gains `detail`, set by `decide` for `InvalidArguments` only. The ledger passes the whole + `Denial` on (`Decided::Denied(Denial)`), and `broker.sock` sends its detail. +3. **`loopd`**: a new `denial_message(reason, detail)` puts the detail after the sentence, before + its full stop. Without a detail it is exactly today's sentence. +4. **Unchanged**: the audit record (`DecisionRecord::Denied { reason }` has no detail: the record + holds the arguments already), `TurnEvent::ToolDenied`, and every text `gatewayd` and `bxctl` + show the owner. Do not touch `crates/gatewayd`, `crates/bxctl` or `crates/bxcore/src/audit.rs`. + +## Every place `ToolResponse::Denied` appears + +This is the complete list. Each one is covered by an edit below or by a given file. + +| Place | What happens there | +|---|---| +| `crates/bxcore/src/wire.rs`, the enum | gains `detail` (edit 1) | +| `crates/brokerd/src/broker.rs`, `handle`, step 3a | sends the `Denial`'s detail (edit B1) | +| `crates/brokerd/src/broker.rs`, `pending`, `Verdict::Denied` and `Err(reason)` | `detail: None`: an approval's denial has no detail (edits B2, B3) | +| `crates/brokerd/src/ledger/answer.rs`, `finish` | a pattern with `..`; no change | +| `crates/loopd/src/turn.rs`, `run_call` | reads `detail` and calls `denial_message` (edit T2) | +| `crates/loopd/src/tools.rs`, `FakeTools::call` | `detail: None` (edit L2) | +| `crates/loopd/src/broker_port.rs` | a pattern with `..`; no change | +| tests in `bxcore`, `brokerd`, `loopd` | new versions are given; `brokerd/tests/broker_sequence.rs` uses `..` and is not changed | + +## Files + +- Copy, from `docs/plans/M5b/files/` (the same path under it): + - new: `crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json`, + `crates/brokerd/tests/invalid_detail.rs` + - **new versions** of: `crates/bxcore/tests/wire.rs`, `crates/bxcore/tests/strict.rs`; + `crates/brokerd/tests/admin.rs`, `broker.rs`, `broker_pending.rs`, `courier.rs`, `ledger.rs`, + `serve.rs`; `crates/loopd/tests/broker_port.rs`, `tools.rs`, `turn_broker.rs` +- Modify: `crates/bxcore/src/wire.rs`, `crates/brokerd/src/args.rs`, + `crates/brokerd/src/policy/mod.rs`, `crates/brokerd/src/ledger/mod.rs`, + `crates/brokerd/src/broker.rs`, `crates/loopd/src/tools.rs`, `crates/loopd/src/turn.rs`, + `docs/implementer-log.md` + +## Facts you need + +- `serde`: `#[serde(default, skip_serializing_if = "Option::is_none")]` on a field of a variant of + this enum (internally tagged, `deny_unknown_fields`) does what it says: a missing `detail` reads + as `None`, `None` is not written, and an unknown key is still refused. Checked against every + fixture; `strict.rs` walks the new one too. +- `char::is_control(self) -> bool` is true for Unicode's control characters only: U+0000 to + U+001F and U+007F to U+009F (tab, line feed, carriage return, escape and NEL among them). It is + false for U+2028 (line separator) and U+2029 (paragraph separator), so `detail` names those two. +- The 300 is characters (`chars()`), not bytes, so a cut never splits a character. +- `str::strip_suffix(&self, '.') -> Option<&str>`; `Option::as_deref(&self) -> Option<&str>` for an + `Option`. +- `Denial { detail: Some(..), ..Denial::new(reason) }` is Rust's struct update syntax: the other + fields come from `Denial::new`. +- `crates/brokerd/src/ledger/mod.rs` already imports `Denial` and `DenyReason`; keep both. + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/bxcore/src/wire.rs` + +**1.** In `enum ToolResponse`, the lines + +```rust + Denied { + reason: DenyReason, + }, +} +``` + +become + +```rust + Denied { + reason: DenyReason, + /// Why the arguments were not valid, for `invalid_arguments` only. Written only when + /// present, so a denial without one has the same bytes as before M5b. + #[serde(default, skip_serializing_if = "Option::is_none")] + detail: Option, + }, +} +``` + +### `crates/brokerd/src/args.rs` + +**A1.** The lines + +```rust +/// Maximum length, in bytes, of a URL. +pub const MAX_URL: usize = 2048; +``` + +become + +```rust +/// Maximum length, in bytes, of a URL. +pub const MAX_URL: usize = 2048; +/// Maximum length, in characters, of the detail an invalid call is denied with. +pub const MAX_DETAIL: usize = 300; +``` + +**A2.** The line + +```rust +impl std::error::Error for ArgsError { +``` + +becomes + +```rust +impl ArgsError { + /// The error as the model is shown it: the `Display` text on one line, each control character + /// and each Unicode line or paragraph separator replaced by a space, cut to `MAX_DETAIL` + /// characters. The text holds the model's own arguments, which may be long or hold newlines. + pub fn detail(&self) -> String { + self.to_string() + .chars() + .map(|c| { + if c.is_control() || c == '\u{2028}' || c == '\u{2029}' { + ' ' + } else { + c + } + }) + .take(MAX_DETAIL) + .collect() + } +} + +impl std::error::Error for ArgsError { +``` + +### `crates/brokerd/src/policy/mod.rs` + +**P1.** In `struct Denial`, the lines + +```rust + /// `Some` only for `DeniedByGrant`: the deny grant's file hash. + pub grant_sha256: Option, +} +``` + +become + +```rust + /// `Some` only for `DeniedByGrant`: the deny grant's file hash. + pub grant_sha256: Option, + /// `Some` only for `InvalidArguments`: why, from `ArgsError::detail`. It goes to the model in + /// the `ToolResponse`, never into the audit record, which holds the arguments already. + pub detail: Option, +} +``` + +**P2.** In `Denial::new`, the lines + +```rust + grant: None, + grant_sha256: None, + } +``` + +become + +```rust + grant: None, + grant_sha256: None, + detail: None, + } +``` + +**P3.** In `decide`, the lines + +```rust + // 2. Malformed arguments are refused before matching, whatever the grants say. + let Ok(args) = args::parse(tool, &request.arguments) else { + return Outcome::Denied(Denial::new(DenyReason::InvalidArguments)); + }; +``` + +become + +```rust + // 2. Malformed arguments are refused before matching, whatever the grants say, and the + // model is told why, so that it can correct the call. + let args = match args::parse(tool, &request.arguments) { + Ok(args) => args, + Err(error) => { + return Outcome::Denied(Denial { + detail: Some(error.detail()), + ..Denial::new(DenyReason::InvalidArguments) + }); + } + }; +``` + +**P4.** In `decide`, the lines + +```rust + Mode::Deny => Outcome::Denied(Denial { + reason: DenyReason::DeniedByGrant, + grant: Some(winner.id.clone()), + grant_sha256: Some(winner.grant_sha256), + }), +``` + +become + +```rust + Mode::Deny => Outcome::Denied(Denial { + reason: DenyReason::DeniedByGrant, + grant: Some(winner.id.clone()), + grant_sha256: Some(winner.grant_sha256), + detail: None, + }), +``` + +**P5.** In `redecide`, the lines + +```rust + Mode::Deny => Err(Denial { + reason: DenyReason::DeniedByGrant, + grant: Some(winner.id.clone()), + grant_sha256: Some(winner.grant_sha256), + }), +``` + +become + +```rust + Mode::Deny => Err(Denial { + reason: DenyReason::DeniedByGrant, + grant: Some(winner.id.clone()), + grant_sha256: Some(winner.grant_sha256), + detail: None, + }), +``` + +### `crates/brokerd/src/ledger/mod.rs` + +**D1.** The lines + +```rust +/// How a call was decided. +#[derive(Debug)] +pub enum Decided { +``` + +become + +```rust +/// How a call was decided. A denial carries the policy's `Denial`, so an invalid call keeps its +/// detail on the way to the answer. +#[derive(Debug)] +pub enum Decided { +``` + +**D2.** At the end of `enum Decided`, the lines + +```rust + Denied(DenyReason), +} +``` + +become + +```rust + Denied(Denial), +} +``` + +**D3.** In `Ledger::decide`, the line + +```rust + None => return Decided::Denied(DenyReason::AuditUnavailable), +``` + +becomes + +```rust + None => return Decided::Denied(Denial::new(DenyReason::AuditUnavailable)), +``` + +**D4.** The line + +```rust + RecordPayload::Simple(reason) => Decided::Denied(reason), +``` + +becomes + +```rust + RecordPayload::Simple(reason) => Decided::Denied(Denial::new(reason)), +``` + +**D5.** The line + +```rust + RecordPayload::Denial(denial) => Decided::Denied(denial.reason), +``` + +becomes + +```rust + RecordPayload::Denial(denial) => Decided::Denied(denial), +``` + +**D6.** The line + +```rust + Err(()) => Decided::Denied(DenyReason::AuditUnavailable), +``` + +becomes + +```rust + Err(()) => Decided::Denied(Denial::new(DenyReason::AuditUnavailable)), +``` + +### `crates/brokerd/src/broker.rs` + +**B1.** In `handle`, the line + +```rust + Decided::Denied(reason) => Some(ToolResponse::Denied { reason }), // 3a. denied +``` + +becomes + +```rust + Decided::Denied(denial) => Some(ToolResponse::Denied { + reason: denial.reason, + detail: denial.detail, + }), // 3a. denied +``` + +**B2.** In `pending`, the line + +```rust + Verdict::Denied(reason) => Some(ToolResponse::Denied { reason }), +``` + +becomes + +```rust + Verdict::Denied(reason) => Some(ToolResponse::Denied { + reason, + detail: None, + }), +``` + +**B3.** At the end of `pending`, the line + +```rust + Err(reason) => Some(ToolResponse::Denied { reason }), +``` + +becomes + +```rust + Err(reason) => Some(ToolResponse::Denied { + reason, + detail: None, + }), +``` + +### `crates/loopd/src/tools.rs` + +**L1.** The line + +```rust +pub struct FakeTools { +``` + +becomes + +```rust +/// What a denial reaches the model as: the reason's sentence, and the broker's detail, if it sent +/// one, after a colon (`Denied: the arguments are not valid for this tool: .`). +pub fn denial_message(reason: bxcore::DenyReason, detail: Option<&str>) -> String { + let sentence = denial_text(reason); + match detail { + None => sentence.to_string(), + Some(detail) => { + let stem = sentence.strip_suffix('.').unwrap_or(sentence); + format!("{stem}: {detail}.") + } + } +} + +pub struct FakeTools { +``` + +`denial_text` itself does not change. + +**L2.** In `FakeTools::call`, the lines + +```rust + _ => bxcore::ToolResponse::Denied { + reason: bxcore::DenyReason::NoGrant, + }, +``` + +become + +```rust + _ => bxcore::ToolResponse::Denied { + reason: bxcore::DenyReason::NoGrant, + detail: None, + }, +``` + +### `crates/loopd/src/turn.rs` + +**T1.** The line + +```rust +use crate::tools::{Registry, ToolPort, cap_result, denial_text, dispatch}; +``` + +becomes + +```rust +use crate::tools::{Registry, ToolPort, cap_result, denial_message, dispatch}; +``` + +**T2.** In `run_call`, the lines + +```rust + ToolResponse::Denied { reason } => { + on_event(&TurnEvent::ToolDenied { + name: request.tool.clone(), + reason, + }); + (denial_text(reason).to_string(), DataClass::Public, false) + } +``` + +become + +```rust + ToolResponse::Denied { reason, detail } => { + on_event(&TurnEvent::ToolDenied { + name: request.tool.clone(), + reason, + }); + ( + denial_message(reason, detail.as_deref()), + DataClass::Public, + false, + ) + } +``` + +The event keeps only the reason: the owner's texts do not change. + +## Steps + +- [ ] **1. Copy the given files.** From the repository root, thirteen commands: + + ```sh + cp docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json \ + crates/bxcore/tests/fixtures/wire/ + cp docs/plans/M5b/files/crates/bxcore/tests/wire.rs \ + crates/bxcore/tests/ + cp docs/plans/M5b/files/crates/bxcore/tests/strict.rs \ + crates/bxcore/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/invalid_detail.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/admin.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/broker.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/broker_pending.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/courier.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/ledger.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/serve.rs \ + crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/broker_port.rs \ + crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/tools.rs \ + crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/turn_broker.rs \ + crates/loopd/tests/ + ``` + + Then `git status --short` lists 11 modified files and 2 new ones, all under `crates/*/tests/`. +- [ ] **2. See them fail.** + - `cargo test -p bxcore --test strict`: one test fails, + `envelopes_reject_unknown_keys_at_every_depth`, with + `tool_response_denied_detail.json: fixture must decode`. + - `cargo test -p bxcore --test wire`: does not compile, with two errors `E0559` + (``variant `bxcore::ToolResponse::Denied` has no field named `detail` ``). + - `cargo build --tests --keep-going -p brokerd -p loopd`: ten test targets do not compile: + `admin`, `broker`, `broker_pending`, `courier`, `invalid_detail`, `ledger` and `serve` in + `brokerd`; `broker_port`, `tools` and `turn_broker` in `loopd`. Every error is about a name + this task adds: `E0559` (no field named `detail` on `ToolResponse::Denied`), `E0609` (no field + `detail` on `Denial`), `E0599` (no method named `detail` for `ArgsError`), `E0432` + (unresolved import `MAX_DETAIL` or `denial_message`), or `E0308` (mismatched types, in + `ledger.rs`, where `Decided::Denied` will hold a `Denial`). +- [ ] **3. `bxcore`.** Make edit 1. `cargo test -p bxcore`: every suite passes. +- [ ] **4. `brokerd`.** Make edits A1, A2, P1 to P5, D1 to D6, and B1 to B3. + `cargo test -p brokerd`: every suite passes, among them the eight tests of `invalid_detail`. +- [ ] **5. `loopd`.** Make edits L1, L2, T1 and T2. `cargo test -p loopd`: every suite passes. +- [ ] **6. Check.** `grep -rn "Decided::Denied(DenyReason" crates/` prints nothing, and + `git diff --stat -- crates/gatewayd crates/bxctl crates/bxcore/src/audit.rs` prints nothing. +- [ ] **7. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 956 tests: 944 before this task, plus 12 new ones (two in + `bxcore/tests/wire.rs`, eight in `brokerd/tests/invalid_detail.rs`, one in + `loopd/tests/tools.rs`, one in `loopd/tests/turn_broker.rs`). +- [ ] **8. Log and commit.** + `git add crates/bxcore crates/brokerd crates/loopd 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 956 tests. +- `grep -rn "Decided::Denied(DenyReason" crates/` prints nothing. + +## Stop and report if + +- Any quoted "before" block is not in its file exactly as shown, or is there more than once. +- Step 2 shows a different result: another test failing in `strict`, or an error of another kind. +- After step 3, 4 or 5, a test fails. Do not change a test or a fixture. +- A test in `gatewayd`, `bxctl`, `memoryd` or `toolkit` 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 956. +- A comment in an edit does not fit the code under it. Quote it and stop. diff --git a/docs/plans/M5b/files/crates/brokerd/tests/admin.rs b/docs/plans/M5b/files/crates/brokerd/tests/admin.rs new file mode 100644 index 0000000..7af5e87 --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/admin.rs @@ -0,0 +1,388 @@ +//! One `admin.sock` connection: list, approve, refuse, check the grants; and expiry, which +//! answers an approval the same way. 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::os::unix::net::UnixStream; +use std::sync::{Arc, Barrier}; +use std::time::{Duration, Instant}; + +use brokerd::admin; +use brokerd::broker::{self, Broker}; +use brokerd::policy::Label; +use bxcore::{ + ApprovalAnswer, Approve, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, ErrorCode, + Message, Refuse, SessionId, Timestamp, ToolResponse, +}; +use client::{Serve, next, open}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +fn ask_notes(rig: &Rig) { + rig.grant( + "notes", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); +} + +/// Sends a call that waits for approval; returns its connection once the pending frame is read. +fn pending(broker: &Arc, call: u64, arguments: &str) -> (UnixStream, u64) { + let req = request("s1", call, "read_file", arguments); + let mut stream = open(broker, broker::handle, call, Message::ToolRequest(req)); + match next(&mut stream).msg { + Message::ToolResponse(ToolResponse::PendingApproval { approval, .. }) => (stream, approval), + other => panic!("{other:?}"), + } +} + +/// One admin request; returns the answer's message after checking its id and `final`. +fn admin(broker: &Arc, msg: Message) -> Message { + let mut stream = open(broker, admin::handle, 5, msg); + let env = next(&mut stream); + assert_eq!((env.id, env.r#final), (5, true), "{env:?}"); + env.msg +} + +fn approve(broker: &Arc, approval: u64) -> Message { + admin(broker, Message::Approve(Approve { approval })) +} + +fn refuse(broker: &Arc, approval: u64, reason: Option<&str>) -> Message { + let reason = reason.map(str::to_string); + admin(broker, Message::Refuse(Refuse { approval, reason })) +} + +fn final_answer(stream: &mut UnixStream) -> ToolResponse { + match next(stream).msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +fn outcome(msg: Message) -> DecisionRecord { + match msg { + Message::ApproveResult(r) => r.outcome, + other => panic!("{other:?}"), + } +} + +fn error_code(msg: &Message) -> Option { + match msg { + Message::Error(e) => Some(e.code), + _ => None, + } +} + +fn denied(reason: DenyReason) -> ToolResponse { + ToolResponse::Denied { + reason, + detail: None, + } +} + +#[test] +fn approvals_lists_the_arguments_as_policy_parsed_them() { + let rig = Rig::new("admin-list"); + ask_notes(&rig); + let broker = rig.broker(&Recording::answering("x")); + assert!(matches!( + admin(&broker, Message::Approvals(Empty {})), + Message::ApprovalList(list) if list.items.is_empty() + )); + // A backslash-u escape of `/` is `/` in JSON; the owner sees the one spelling policy matched. + let escaped = format!(r#"{{"path":"{}n{}a"}}"#, "\\u002f", "\\u002f"); + assert!( + escaped.contains("u002f"), + "the escape must survive: {escaped}" + ); + let (_stream, approval) = pending(&broker, 1, &escaped); + match admin(&broker, Message::Approvals(Empty {})) { + Message::ApprovalList(list) => { + assert_eq!(list.items.len(), 1); + assert_eq!(list.items[0].approval, approval); + assert_eq!(list.items[0].arguments, r#"{"path":"/n/a"}"#); + assert_eq!(list.items[0].grant, "notes"); + } + other => panic!("{other:?}"), + } + // The audit log keeps the raw string. + match &rig.events()[0] { + AuditEvent::Decision { arguments, .. } => assert_eq!(arguments, &escaped), + other => panic!("{other:?}"), + } +} + +#[test] +fn approve_runs_the_call_and_answers_with_the_re_decision() { + let rig = Rig::new("admin-approve"); + ask_notes(&rig); + let rt = Recording::answering("the notes"); + let broker = rig.broker(&rt); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + assert_eq!(outcome(approve(&broker, approval)), DecisionRecord::Ask {}); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert_eq!(rt.count(), 1); + match &rig.events()[1] { + AuditEvent::Approval { answer, by, .. } => { + assert_eq!(*answer, ApprovalAnswer::Approved); + assert_eq!(by.as_deref(), Some("bxctl")); + } + other => panic!("{other:?}"), + } +} + +#[test] +fn refuse_denies_the_call_and_records_the_reason() { + let rig = Rig::new("admin-refuse"); + ask_notes(&rig); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + assert!(matches!( + refuse(&broker, approval, Some("not today")), + Message::Ok(Empty {}) + )); + assert_eq!( + final_answer(&mut stream), + denied(DenyReason::ApprovalRefused) + ); + assert_eq!(rt.count(), 0); + match &rig.events()[1] { + AuditEvent::Approval { answer, reason, .. } => { + assert_eq!(*answer, ApprovalAnswer::Refused); + assert_eq!(reason.as_deref(), Some("not today")); + } + other => panic!("{other:?}"), + } +} + +#[test] +fn an_unknown_or_answered_id_is_no_such_approval() { + let rig = Rig::new("admin-unknown"); + ask_notes(&rig); + let broker = rig.broker(&Recording::answering("x")); + assert_eq!( + error_code(&approve(&broker, 41)), + Some(ErrorCode::NoSuchApproval) + ); + assert_eq!( + error_code(&refuse(&broker, 41, None)), + Some(ErrorCode::NoSuchApproval) + ); + let (_stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + refuse(&broker, approval, None); + assert_eq!( + error_code(&approve(&broker, approval)), + Some(ErrorCode::NoSuchApproval) + ); + assert_eq!(rig.events().len(), 2, "one decision, one approval"); +} + +#[test] +fn approve_and_refuse_at_once_give_exactly_one_answer() { + let rig = Rig::new("admin-race"); + ask_notes(&rig); + let broker = rig.broker(&Recording::answering("x")); + for round in 0..100u64 { + let (mut stream, approval) = pending(&broker, round, r#"{"path":"/n/a"}"#); + let start = Arc::new(Barrier::new(2)); + let racers: Vec<_> = [true, false] + .into_iter() + .map(|approving| { + let (broker, start) = (Arc::clone(&broker), Arc::clone(&start)); + std::thread::spawn(move || { + start.wait(); + if approving { + approve(&broker, approval) + } else { + refuse(&broker, approval, None) + } + }) + }) + .collect(); + let answers: Vec = racers.into_iter().map(|r| r.join().unwrap()).collect(); + let losers = answers + .iter() + .filter(|m| error_code(m) == Some(ErrorCode::NoSuchApproval)) + .count(); + assert_eq!(losers, 1, "round {round}: {answers:?}"); + final_answer(&mut stream); + let approvals = rig + .events() + .iter() + .filter(|e| matches!(e, AuditEvent::Approval { decision, .. } if *decision == approval)) + .count(); + assert_eq!(approvals, 1, "round {round}"); + } +} + +#[test] +fn approve_after_the_grant_file_is_removed_is_denied() { + let rig = Rig::new("admin-removed"); + ask_notes(&rig); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + rig.remove_grant("notes"); + let reason = DenyReason::NoGrant; + assert_eq!( + outcome(approve(&broker, approval)), + DecisionRecord::Denied { reason } + ); + assert_eq!(final_answer(&mut stream), denied(reason)); + assert_eq!(rt.count(), 0); +} + +#[test] +fn approve_after_the_taint_rose_past_the_grant_is_denied() { + let rig = Rig::new("admin-taint"); + let text = grant_text("read_file", "ask", "", "paths = [\"/n\"]") + .replace("max_taint = \"secret\"", "max_taint = \"private\""); + rig.grant("notes", &text); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + let s1 = SessionId::new("s1").unwrap(); + let secret = Label { + class: DataClass::Secret, + untrusted: false, + }; + rig.state() + .raise(&s1, rig.state().read(&s1).unwrap(), secret) + .unwrap(); + let reason = DenyReason::TaintTooHigh; + assert_eq!( + outcome(approve(&broker, approval)), + DecisionRecord::Denied { reason } + ); + assert_eq!(final_answer(&mut stream), denied(reason)); + assert_eq!(rt.count(), 0); +} + +#[test] +fn an_approval_that_cannot_be_recorded_is_denied_on_both_sides() { + let rig = Rig::new("admin-norecord"); + ask_notes(&rig); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + rig.switch.fail(true); + let reason = DenyReason::AuditUnavailable; + assert_eq!( + outcome(approve(&broker, approval)), + DecisionRecord::Denied { reason } + ); + assert_eq!(final_answer(&mut stream), denied(reason)); + assert_eq!(rt.count(), 0); +} + +#[test] +fn a_refusal_that_cannot_be_recorded_is_an_error_for_bxctl() { + let rig = Rig::new("admin-norefuse"); + ask_notes(&rig); + let broker = rig.broker(&Recording::answering("x")); + let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + rig.switch.fail(true); + match refuse(&broker, approval, None) { + Message::Error(e) => { + assert_eq!(e.code, ErrorCode::Internal); + assert!( + e.detail.ends_with("see docs/runbook.md#audit-unavailable"), + "{}", + e.detail + ); + } + other => panic!("{other:?}"), + } + assert_eq!( + final_answer(&mut stream), + denied(DenyReason::AuditUnavailable) + ); +} + +#[test] +fn expiry_denies_an_approval_nobody_answered() { + let rig = Rig::with_ttl("admin-expire", 100); + ask_notes(&rig); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, _) = pending(&broker, 1, r#"{"path":"/n/a"}"#); + let until = Instant::now() + Duration::from_secs(10); + while admin::expire_due(&broker, Timestamp::now()) == 0 { + assert!(Instant::now() < until, "never expired"); + std::thread::sleep(Duration::from_millis(20)); + } + assert_eq!( + final_answer(&mut stream), + denied(DenyReason::ApprovalExpired) + ); + match &rig.events()[1] { + AuditEvent::Approval { answer, by, .. } => { + assert_eq!((*answer, by.as_deref()), (ApprovalAnswer::Expired, None)); + } + other => panic!("{other:?}"), + } + assert_eq!(rt.count(), 0); +} + +#[test] +fn check_grants_reports_every_problem_or_none() { + let rig = Rig::new("admin-grants"); + ask_notes(&rig); + let broker = rig.broker(&Recording::answering("x")); + match admin(&broker, Message::CheckGrants(Empty {})) { + Message::GrantsReport(r) => assert!(r.problems.is_empty(), "{r:?}"), + other => panic!("{other:?}"), + } + rig.grant("bad", "tool = \"read_file\"\nmdoe = \"auto\"\n"); + rig.grant( + "Worse", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); + match admin(&broker, Message::CheckGrants(Empty {})) { + Message::GrantsReport(r) => { + let files: Vec<&str> = r.problems.iter().map(|p| p.file.as_str()).collect(); + assert!(files.contains(&"bad.toml"), "{r:?}"); + assert!(files.contains(&"Worse.toml"), "{r:?}"); + } + other => panic!("{other:?}"), + } +} + +#[test] +fn every_other_kind_on_admin_sock_is_forbidden() { + let rig = Rig::new("admin-forbidden"); + let broker = rig.broker(&Recording::answering("x")); + let req = request("s1", 1, "read_file", r#"{"path":"/n/a"}"#); + for msg in [ + Message::ToolRequest(req), + Message::Ok(Empty {}), + Message::ToolResponse(denied(DenyReason::NoGrant)), + ] { + assert_eq!(error_code(&admin(&broker, msg)), Some(ErrorCode::Forbidden)); + } + let lines = rig.lines.with("on admin.sock"); + assert_eq!(lines.len(), 3, "{:?}", rig.lines.all()); + assert!(lines[0].contains("tool_request"), "{}", lines[0]); + assert!( + lines + .iter() + .all(|l| l.ends_with("see docs/runbook.md#socket-forbidden")) + ); + assert!( + rig.events().is_empty(), + "a refused tool request decides nothing" + ); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/broker.rs b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs new file mode 100644 index 0000000..fc31dcb --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs @@ -0,0 +1,327 @@ +//! One `broker.sock` connection: decision, record, run, answer; and every way it fails closed. +//! 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::fs::PermissionsExt; +use std::os::unix::net::UnixStream; +use std::sync::Arc; + +use brokerd::broker; +use brokerd::ledger::NOT_RECORDED; +use brokerd::runner::{REFUSING, RunError}; +use bxcore::{ + ApprovalList, Approve, ApproveResult, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, + Envelope, ErrorCode, GrantsReport, Message, Refuse, ResultStatus, SessionId, ToolResponse, + Turn, TurnDone, TurnEvent, Usage, WireError, +}; +use client::{Serve, call, next, open}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +const NOTES: &str = r#"{"path":"/n/a"}"#; + +fn read_notes(call_id: u64) -> bxcore::ToolRequest { + request("s1", call_id, "read_file", NOTES) +} + +fn only(frames: &[Envelope]) -> &ToolResponse { + assert_eq!(frames.len(), 1, "{frames:?}"); + assert!(frames[0].r#final); + match &frames[0].msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +fn denied(reason: DenyReason) -> ToolResponse { + ToolResponse::Denied { + reason, + detail: None, + } +} + +fn auto_notes(rig: &Rig, class: &str) { + let text = grant_text("read_file", "auto", "", "paths = [\"/n\"]").replace( + "result_class = \"private\"", + &format!("result_class = \"{class}\""), + ); + rig.grant("notes", &text); +} + +#[test] +fn a_call_no_grant_allows_is_denied_and_recorded() { + let rig = Rig::new("broker-nogrant"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(41)); + assert_eq!(frames[0].id, 41, "the answer carries the request's id"); + assert_eq!(only(&frames), &denied(DenyReason::NoGrant)); + assert_eq!(rt.count(), 0); + match rig.events().as_slice() { + [AuditEvent::Decision { outcome, .. }] => assert_eq!( + *outcome, + DecisionRecord::Denied { + reason: DenyReason::NoGrant + } + ), + other => panic!("{other:?}"), + } +} + +#[test] +fn an_allowed_call_runs_and_its_result_raises_the_state() { + let rig = Rig::new("broker-allowed"); + auto_notes(&rig, "secret"); + let rt = Recording::answering("the notes"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(7)); + assert_eq!( + only(&frames), + &ToolResponse::Result { + content: "the notes".to_string(), + class: DataClass::Secret, + untrusted: false, + truncated: false, + } + ); + assert_eq!(rt.count(), 1); + let events = rig.events(); + assert!(matches!( + events.as_slice(), + [ + AuditEvent::Decision { .. }, + AuditEvent::Result { + decision: 0, + status: ResultStatus::Result, + taint_after: DataClass::Secret, + .. + } + ] + )); + let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap(); + assert_eq!(state.taint, DataClass::Secret); +} + +#[test] +fn a_runtime_failure_is_passed_on_and_recorded_as_failed() { + let rig = Rig::new("broker-refusing"); + auto_notes(&rig, "secret"); + let rt = Recording::with(Err(RunError::Unavailable(REFUSING.to_string()))); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + assert_eq!( + only(&frames), + &ToolResponse::Failed { + message: REFUSING.to_string() + } + ); + match &rig.events()[1] { + AuditEvent::Result { + status, + taint_after, + .. + } => assert_eq!( + (*status, *taint_after), + (ResultStatus::Failed, DataClass::Private) + ), + other => panic!("{other:?}"), + } + assert!(!rig.state_file("s1").exists(), "a failure changes no state"); +} + +fn every_kind_but_tool_request() -> Vec { + let usage = Usage { + cache_n: 0, + prompt_n: 0, + predicted_n: 0, + reasoning_tokens: 0, + thinking_capped: false, + }; + vec![ + Message::ToolResponse(denied(DenyReason::NoGrant)), + Message::Error(WireError { + code: ErrorCode::Internal, + detail: String::new(), + }), + Message::Turn(Turn { + session: SessionId::new("s1").unwrap(), + content: String::new(), + resume: false, + }), + Message::TurnEvent(TurnEvent::Content { + text: String::new(), + }), + Message::TurnDone(TurnDone { + content: String::new(), + usage, + }), + Message::Approvals(Empty {}), + Message::ApprovalList(ApprovalList { items: Vec::new() }), + Message::Approve(Approve { approval: 0 }), + Message::ApproveResult(ApproveResult { + outcome: DecisionRecord::Allowed {}, + }), + Message::Refuse(Refuse { + approval: 0, + reason: None, + }), + Message::Ok(Empty {}), + Message::CheckGrants(Empty {}), + Message::GrantsReport(GrantsReport { + problems: Vec::new(), + }), + ] +} + +#[test] +fn every_other_kind_on_broker_sock_is_forbidden() { + let rig = Rig::new("broker-forbidden"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let kinds = every_kind_but_tool_request(); + assert_eq!(kinds.len(), 13, "every Message variant but tool_request"); + for (i, msg) in kinds.into_iter().enumerate() { + let id = 100 + i as u64; + let mut stream = open(&broker, broker::handle, id, msg); + let env = next(&mut stream); + assert_eq!((env.id, env.r#final), (id, true)); + match env.msg { + Message::Error(e) => assert_eq!(e.code, ErrorCode::Forbidden), + other => panic!("{other:?}"), + } + } + let lines = rig.lines.with("broker.sock"); + assert_eq!(lines.len(), 13, "{:?}", rig.lines.all()); + for line in &lines { + assert!( + line.ends_with("\nsee docs/runbook.md#socket-forbidden"), + "{line}" + ); + } + assert_eq!(rig.lines.with("kind approve on broker.sock").len(), 1); + assert!( + rig.events().is_empty(), + "nothing is recorded for a refused kind" + ); +} + +#[test] +fn a_frame_that_is_not_json_is_answered_with_bad_message() { + let rig = Rig::new("broker-badframe"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut client, server) = UnixStream::pair().unwrap(); + let b = Arc::clone(&broker); + std::thread::spawn(move || broker::handle(server, &b)); + client.write_all(&3u32.to_be_bytes()).unwrap(); + client.write_all(b"{{{").unwrap(); + let env = next(&mut client); + assert_eq!((env.id, env.r#final), (0, true)); + match env.msg { + Message::Error(e) => assert_eq!(e.code, ErrorCode::BadMessage), + other => panic!("{other:?}"), + } + assert!(rig.events().is_empty()); +} + +#[test] +fn one_invalid_grant_file_denies_a_call_a_valid_file_would_allow() { + let rig = Rig::new("broker-invalid"); + auto_notes(&rig, "private"); + rig.grant("bad", "tool = \"read_file\"\nmdoe = \"auto\"\n"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + for id in [1, 2] { + let frames = call(&broker, read_notes(id)); + assert_eq!(only(&frames), &denied(DenyReason::GrantsInvalid)); + } + assert_eq!(rt.count(), 0); + // Printed once for the two calls, with the pointer. + let printed = rig.lines.with("bad.toml"); + assert_eq!(printed.len(), 1, "{:?}", rig.lines.all()); + assert!( + printed[0].ends_with("see docs/runbook.md#grants-invalid"), + "{}", + printed[0] + ); + rig.remove_grant("bad"); + let frames = call(&broker, read_notes(3)); + assert!(matches!(only(&frames), ToolResponse::Result { .. })); + assert_eq!(rt.count(), 1); +} + +#[test] +fn a_damaged_session_state_is_state_unreadable() { + let rig = Rig::new("broker-state"); + auto_notes(&rig, "private"); + std::fs::create_dir_all(rig.cfg.state_dir()).unwrap(); + std::fs::write(rig.state_file("s1"), "{\"taint\":\"loud\"}").unwrap(); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + assert_eq!(only(&frames), &denied(DenyReason::StateUnreadable)); + assert_eq!(rt.count(), 0); + let lines = rig.lines.with("s1.json"); + assert!( + lines + .iter() + .any(|l| l.ends_with("see docs/runbook.md#broker-state-damaged")), + "{lines:?}" + ); +} + +#[test] +fn a_failed_audit_write_runs_nothing_now_or_later() { + let rig = Rig::new("broker-audit"); + auto_notes(&rig, "private"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + rig.switch.fail(true); + let frames = call(&broker, read_notes(1)); + assert_eq!(only(&frames), &denied(DenyReason::AuditUnavailable)); + rig.switch.fail(false); + let frames = call(&broker, read_notes(2)); + assert_eq!(only(&frames), &denied(DenyReason::AuditUnavailable)); + assert_eq!(rt.count(), 0); + assert!( + !rig.lines + .with("see docs/runbook.md#audit-unavailable") + .is_empty() + ); +} + +#[test] +fn a_state_that_cannot_be_written_withholds_the_content() { + if tmp::running_as_root("a_state_that_cannot_be_written_withholds_the_content") { + return; + } + let rig = Rig::new("broker-ro"); + auto_notes(&rig, "secret"); + let dir = rig.cfg.state_dir(); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let rt = Recording::answering("the secret"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!( + only(&frames), + &ToolResponse::Failed { + message: NOT_RECORDED.to_string() + } + ); + assert_eq!(rt.count(), 1, "it ran; its content is what is withheld"); + let text = format!("{frames:?}"); + assert!(!text.contains("the secret"), "{text}"); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/broker_pending.rs b/docs/plans/M5b/files/crates/brokerd/tests/broker_pending.rs new file mode 100644 index 0000000..81ea335 --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/broker_pending.rs @@ -0,0 +1,233 @@ +//! A call an `ask` grant matched: the pending frame, the table entry, the wait, and a requester +//! that goes away. The tests answer entries by hand, as `admin` will. 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::os::unix::net::UnixStream; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use brokerd::broker::{self, Broker, GONE}; +use brokerd::ledger::Answer; +use bxcore::{ + AuditEvent, DataClass, DecisionRecord, DenyReason, Message, ResultStatus, Timestamp, + ToolResponse, +}; +use client::{Serve, next, open}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +fn ask_notes(rig: &Rig, extra: &str) { + rig.grant( + "notes", + &grant_text("read_file", "ask", extra, "paths = [\"/n\"]"), + ); +} + +/// Sends a call and reads its pending frame. Returns the connection and the frame's values. +fn start(broker: &Arc, arguments: &str) -> (UnixStream, u64, Timestamp) { + let req = request("s1", 9, "read_file", arguments); + let mut stream = open(broker, broker::handle, 9, Message::ToolRequest(req)); + let env = next(&mut stream); + assert_eq!((env.id, env.r#final), (9, false), "{env:?}"); + match env.msg { + Message::ToolResponse(ToolResponse::PendingApproval { approval, expires }) => { + (stream, approval, expires) + } + other => panic!("{other:?}"), + } +} + +/// What `admin` does for an entry it took: record the answer, send the verdict. +fn answer_by_hand(broker: &Broker, id: u64, answer: Answer) -> DecisionRecord { + let entry = broker.table().take(id).expect("the entry is pending"); + answer_entry(broker, entry, answer) +} + +fn answer_entry( + broker: &Broker, + entry: brokerd::approvals::Entry, + answer: Answer, +) -> DecisionRecord { + let grants = broker.grants(); + let done = broker.ledger().answer( + entry.ask, + entry.info.approval, + answer, + &grants, + Timestamp::now(), + ); + entry.reply.send(done.verdict).unwrap(); + done.outcome +} + +fn approved() -> Answer { + Answer::Approved { + by: Some("bxctl".to_string()), + post: None, + } +} + +fn final_answer(stream: &mut UnixStream) -> ToolResponse { + let env = next(stream); + assert!(env.r#final, "{env:?}"); + match env.msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +fn eventually(what: &str, mut done: impl FnMut() -> bool) { + let until = Instant::now() + Duration::from_secs(10); + while !done() { + assert!(Instant::now() < until, "never happened: {what}"); + std::thread::sleep(Duration::from_millis(20)); + } +} + +#[test] +fn an_ask_call_waits_is_listed_and_runs_when_approved() { + let rig = Rig::new("pending-approve"); + ask_notes(&rig, ""); + let rt = Recording::answering("the notes"); + let broker = rig.broker(&rt); + let before = Timestamp::now(); + // Spaced JSON: the table shows the arguments as policy parsed them, not as sent. + let (mut stream, approval, expires) = start(&broker, r#"{ "path" : "/n/a" }"#); + assert_eq!(approval, 0, "the approval id is the decision's seq"); + let items = broker.table().list(); + assert_eq!(items.len(), 1); + let item = &items[0]; + assert_eq!(item.approval, 0); + assert_eq!((item.session.as_str(), item.call.0), ("s1", 9)); + assert_eq!(item.tool, "read_file"); + assert_eq!(item.arguments, r#"{"path":"/n/a"}"#); + assert_eq!( + (item.grant.as_str(), item.taint), + ("notes", DataClass::Private) + ); + assert!(item.created >= before); + assert_eq!(item.expires, expires); + assert_eq!( + expires.unix_millis() - item.created.unix_millis(), + 900_000, + "now plus ttl_ms" + ); + assert_eq!(rt.count(), 0, "nothing runs while it waits"); + + assert_eq!( + answer_by_hand(&broker, 0, approved()), + DecisionRecord::Ask {} + ); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert_eq!(rt.count(), 1); + let events = rig.events(); + assert!( + matches!( + events.as_slice(), + [ + AuditEvent::Decision { .. }, + AuditEvent::Approval { decision: 0, .. }, + AuditEvent::Result { decision: 0, .. } + ] + ), + "{events:?}" + ); +} + +#[test] +fn the_expiry_is_the_grants_when_that_is_earlier() { + let rig = Rig::new("pending-grant-expiry"); + let soon = Timestamp::from_unix_millis(Timestamp::now().unix_millis() + 60_000).unwrap(); + ask_notes(&rig, &format!("expires = \"{}\"", soon.to_rfc3339())); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (_stream, _, expires) = start(&broker, r#"{"path":"/n/a"}"#); + assert_eq!(expires, soon); +} + +#[test] +fn a_denied_verdict_reaches_the_requester_and_nothing_runs() { + let rig = Rig::new("pending-refuse"); + ask_notes(&rig, ""); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, approval, _) = start(&broker, r#"{"path":"/n/a"}"#); + let refused = Answer::Refused { + by: Some("bxctl".to_string()), + post: None, + reason: None, + }; + answer_by_hand(&broker, approval, refused); + assert_eq!( + final_answer(&mut stream), + ToolResponse::Denied { + reason: DenyReason::ApprovalRefused, + detail: None, + } + ); + assert_eq!(rt.count(), 0); +} + +#[test] +fn a_requester_that_leaves_takes_its_own_entry_and_nothing_is_written() { + let rig = Rig::new("pending-leave"); + ask_notes(&rig, ""); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (stream, _, _) = start(&broker, r#"{"path":"/n/a"}"#); + drop(stream); + // The waiting thread looks at its connection every second. + eventually("the entry is removed", || broker.table().list().is_empty()); + std::thread::sleep(Duration::from_millis(200)); + assert_eq!(rig.events().len(), 1, "only the decision"); + let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap(); + assert_eq!(report.abandoned, [0]); + assert!(report.failure.is_none()); + assert_eq!(rt.count(), 0); +} + +#[test] +fn a_requester_gone_at_the_last_look_runs_nothing_and_closes_the_call() { + let rig = Rig::new("pending-lastlook"); + ask_notes(&rig, ""); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (stream, approval, _) = start(&broker, r#"{"path":"/n/a"}"#); + // Take the entry first, so the waiting thread cannot take it back when it sees the + // connection gone; then approve. + let entry = broker.table().take(approval).unwrap(); + drop(stream); + answer_entry(&broker, entry, approved()); + eventually("the call is closed", || rig.events().len() == 3); + match &rig.events()[2] { + AuditEvent::Result { + status, + decision, + sha256, + bytes, + .. + } => { + assert_eq!((*status, *decision), (ResultStatus::Failed, 0)); + assert_eq!(*sha256, bxcore::sha256(GONE.as_bytes()).unwrap()); + assert_eq!(*bytes, GONE.len() as u64); + } + other => panic!("{other:?}"), + } + assert_eq!(GONE, "the requester went away"); + assert_eq!(rt.count(), 0); + let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap(); + assert!(report.unfinished.is_empty() && report.abandoned.is_empty()); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/courier.rs b/docs/plans/M5b/files/crates/brokerd/tests/courier.rs new file mode 100644 index 0000000..625de1f --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/courier.rs @@ -0,0 +1,449 @@ +//! The courier carries a pending approval to `gatewayd` and its answer back, against a fake +//! `gatewayd` on a real socket (M4b spec, section 5). Do not edit. + +#[path = "support/client.rs"] +mod client; +#[path = "support/fake_gateway.rs"] +mod fake_gateway; +#[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::os::unix::net::UnixStream; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::time::{Duration, Instant}; + +use brokerd::broker::{self, Broker}; +use brokerd::config::Gateway; +use brokerd::courier::{self, Timing}; +use brokerd::ledger::Answer; +use bxcore::{ + ApprovalAnswer, ApprovalReply, AuditEvent, DenyReason, GatewayOutcome, Message, Timestamp, + ToolResponse, +}; +use client::{Serve, next, open}; +use fake_gateway::{FakeGateway, KYLE, POST}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +const WAIT: Duration = Duration::from_secs(5); + +/// A rig whose `[gateway]` is a fake gatewayd in its home, and an `ask` grant for /n. +fn rig_with_gateway(tag: &str) -> (Rig, FakeGateway) { + let mut rig = Rig::new(tag); + let socket = rig.dir.path().join("gw.sock"); + let fake = FakeGateway::start(&socket); + rig.cfg.gateway = Some(Gateway { socket }); + rig.grant( + "notes", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); + (rig, fake) +} + +/// A call that waits for approval; its connection and the approval id. +fn ask(broker: &Arc) -> (UnixStream, u64) { + let mut stream = open( + broker, + broker::handle, + 9, + Message::ToolRequest(request("s1", 9, "read_file", r#"{"path":"/n/a"}"#)), + ); + match next(&mut stream).msg { + Message::ToolResponse(ToolResponse::PendingApproval { approval, .. }) => (stream, approval), + other => panic!("{other:?}"), + } +} + +fn final_answer(stream: &mut UnixStream) -> ToolResponse { + let env = next(stream); + assert!(env.r#final, "{env:?}"); + match env.msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +/// The `Approval` record's (answer, by, post, reason). +fn approval_record( + rig: &Rig, +) -> ( + ApprovalAnswer, + Option, + Option, + Option, +) { + let events = rig.events(); + let found = events.iter().find_map(|e| match e { + AuditEvent::Approval { + answer, + by, + post, + reason, + .. + } => Some((*answer, by.clone(), post.clone(), reason.clone())), + _ => None, + }); + found.expect("an Approval record") +} + +fn some(s: &str) -> Option { + Some(s.to_string()) +} + +/// Sets the courier's stop flag when dropped, so a failing assertion ends the courier too. +struct StopOnDrop<'a>(&'a AtomicBool); + +impl Drop for StopOnDrop<'_> { + fn drop(&mut self) { + self.0.store(true, std::sync::atomic::Ordering::SeqCst); + } +} + +/// The courier must end by itself within five seconds. +fn ends_by_itself(courier: &std::thread::ScopedJoinHandle<'_, T>) { + let until = Instant::now() + WAIT; + while !courier.is_finished() { + assert!(Instant::now() < until, "the courier did not stop"); + std::thread::sleep(Duration::from_millis(20)); + } +} + +#[test] +fn approved_in_mattermost_it_runs_and_the_record_names_who_and_where() { + let (rig, fake) = rig_with_gateway("courier-approve"); + let rt = Recording::answering("the notes"); + let broker = rig.broker(&rt); + let (mut stream, id) = ask(&broker); + let mut conn = fake.next(WAIT).expect("the approval reaches gatewayd"); + assert_eq!( + (conn.request.approval, conn.request.tool.as_str()), + (id, "read_file") + ); + assert_eq!(conn.request.arguments, r#"{"path":"/n/a"}"#); + conn.posted(); + conn.approve(); + assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {})); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert_eq!(rt.count(), 1); + assert_eq!( + approval_record(&rig), + (ApprovalAnswer::Approved, some(KYLE), some(POST), None) + ); +} + +#[test] +fn refused_in_mattermost_it_is_denied_with_the_reason() { + let (rig, fake) = rig_with_gateway("courier-refuse"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, _) = ask(&broker); + let mut conn = fake.next(WAIT).unwrap(); + conn.posted(); + conn.refuse("not on a Friday"); + assert_eq!(conn.outcome(), Some(GatewayOutcome::Refused {})); + assert_eq!( + final_answer(&mut stream), + ToolResponse::Denied { + reason: DenyReason::ApprovalRefused, + detail: None, + } + ); + assert_eq!(rt.count(), 0); + assert_eq!( + approval_record(&rig), + ( + ApprovalAnswer::Refused, + some(KYLE), + some(POST), + some("not on a Friday") + ) + ); +} + +#[test] +fn approved_after_the_grant_is_gone_it_is_denied_on_the_recheck() { + let (rig, fake) = rig_with_gateway("courier-recheck"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, _) = ask(&broker); + let mut conn = fake.next(WAIT).unwrap(); + conn.posted(); + rig.remove_grant("notes"); + conn.approve(); + assert_eq!( + conn.outcome(), + Some(GatewayOutcome::Denied { + reason: DenyReason::NoGrant + }) + ); + assert_eq!( + final_answer(&mut stream), + ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + } + ); + assert_eq!(rt.count(), 0); +} + +#[test] +fn answered_in_bxctl_the_courier_closes_without_an_outcome() { + let (rig, fake) = rig_with_gateway("courier-bxctl"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, id) = ask(&broker); + let mut conn = fake.next(WAIT).unwrap(); + conn.posted(); + let entry = broker.table().take(id).unwrap(); + brokerd::admin::answer( + &broker, + entry, + Answer::Approved { + by: some("bxctl"), + post: None, + }, + Timestamp::now(), + ); + assert!( + conn.closed_within(WAIT), + "the courier closes, and sends nothing" + ); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert_eq!(approval_record(&rig).1, some("bxctl")); +} + +#[test] +fn an_answer_after_it_ended_is_gone() { + let rig = Rig::new("courier-gone"); + rig.grant( + "notes", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (_stream, id) = ask(&broker); + // The courier is run by hand here, so that the answer can arrive after the entry is taken. + let socket = rig.dir.path().join("gw.sock"); + let fake = FakeGateway::start(&socket); + let info = broker.table().list().remove(0); + let stop = AtomicBool::new(false); + std::thread::scope(|scope| { + let courier = scope.spawn(|| { + courier::carry( + &broker, + &socket, + &info, + &stop, + Timing { + retry: Duration::from_millis(100), + tick: Duration::from_secs(5), + }, + ) + }); + let _guard = StopOnDrop(&stop); + let mut conn = fake.next(WAIT).unwrap(); + conn.posted(); + let entry = broker.table().take(id).unwrap(); + brokerd::admin::answer(&broker, entry, Answer::Expired, Timestamp::now()); + conn.approve(); + assert_eq!(conn.outcome(), Some(GatewayOutcome::Gone {})); + ends_by_itself(&courier); + }); +} + +#[test] +fn if_gatewayd_goes_away_the_approval_is_posted_again() { + let (rig, fake) = rig_with_gateway("courier-again"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, id) = ask(&broker); + let mut first = fake.next(WAIT).unwrap(); + first.posted(); + first.close(); + let mut second = fake.next(Duration::from_secs(10)).expect("posted again"); + assert_eq!(second.request.approval, id); + second.posted(); + second.approve(); + assert_eq!(second.outcome(), Some(GatewayOutcome::Runs {})); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert!( + rig.lines.all().iter().any(|l| l.contains(&format!( + "approval {id}: gatewayd went away; posting it again" + ))), + "{:?}", + rig.lines.all() + ); +} + +#[test] +fn a_bad_answer_answers_nothing() { + let (rig, fake) = rig_with_gateway("courier-bad"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (_stream, id) = ask(&broker); + for bad in [ + ApprovalReply { + approval: id + 1, + approved: true, + by: KYLE.to_string(), + post: POST.to_string(), + reason: None, + }, + ApprovalReply { + approval: id, + approved: true, + by: "kyle".to_string(), + post: POST.to_string(), + reason: None, + }, + ApprovalReply { + approval: id, + approved: false, + by: KYLE.to_string(), + post: POST.to_string(), + reason: Some("x".repeat(501)), + }, + ] { + let mut conn = fake.next(Duration::from_secs(10)).unwrap(); + conn.posted(); + conn.reply(bad); + assert_eq!(conn.outcome(), None, "no outcome for a bad answer"); + assert!(broker.table().contains(id), "still pending"); + } + assert_eq!(rt.count(), 0); +} + +#[test] +fn unreachable_at_first_then_posted_and_said_once() { + let rig = Rig::new("courier-late"); + rig.grant( + "notes", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (_stream, id) = ask(&broker); + let socket = rig.dir.path().join("gw.sock"); + let info = broker.table().list().remove(0); + let stop = AtomicBool::new(false); + let timing = Timing { + retry: Duration::from_millis(100), + tick: Duration::from_millis(20), + }; + std::thread::scope(|scope| { + let courier = scope.spawn(|| courier::carry(&broker, &socket, &info, &stop, timing)); + let _guard = StopOnDrop(&stop); + std::thread::sleep(Duration::from_millis(500)); + let fake = FakeGateway::start(&socket); + let mut conn = fake.next(WAIT).unwrap(); + conn.posted(); + let until = Instant::now() + WAIT; + while !rig + .lines + .all() + .iter() + .any(|l| l.contains("posted in Mattermost")) + { + assert!(Instant::now() < until, "{:?}", rig.lines.all()); + std::thread::sleep(Duration::from_millis(20)); + } + broker.table().take(id); + ends_by_itself(&courier); + }); + let lines = rig.lines.all(); + let unreachable: Vec<&String> = lines + .iter() + .filter(|l| l.contains("gatewayd is not reachable")) + .collect(); + assert_eq!( + unreachable.len(), + 1, + "said once, not every retry: {lines:?}" + ); + assert!( + unreachable[0].starts_with(&format!( + "brokerd: approval {id}: gatewayd is not reachable at " + )), + "{lines:?}" + ); + assert!( + unreachable[0] + .ends_with("; bxctl can still answer it\nsee docs/runbook.md#gateway-unreachable"), + "{lines:?}" + ); + assert!( + lines.contains(&format!("brokerd: approval {id} posted in Mattermost")), + "{lines:?}" + ); +} + +#[test] +fn if_gatewayd_cannot_post_it_is_tried_again() { + let (rig, fake) = rig_with_gateway("courier-cannot"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, _) = ask(&broker); + fake.next(WAIT).unwrap().cannot_post(); + let mut conn = fake.next(Duration::from_secs(10)).expect("tried again"); + conn.posted(); + conn.approve(); + assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {})); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); + assert!( + rig.lines + .all() + .iter() + .any(|l| l.contains("gatewayd could not post it: Mattermost is down")) + ); +} + +#[test] +fn without_a_gateway_nothing_is_carried() { + let rig = Rig::new("courier-none"); + rig.grant( + "notes", + &grant_text("read_file", "ask", "", "paths = [\"/n\"]"), + ); + let socket = rig.dir.path().join("gw.sock"); + let fake = FakeGateway::start(&socket); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut stream, id) = ask(&broker); + assert!( + fake.next(Duration::from_millis(500)).is_none(), + "no [gateway]: no connection" + ); + let entry = broker.table().take(id).unwrap(); + brokerd::admin::answer( + &broker, + entry, + Answer::Approved { + by: some("bxctl"), + post: None, + }, + Timestamp::now(), + ); + assert!(matches!( + final_answer(&mut stream), + ToolResponse::Result { .. } + )); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/invalid_detail.rs b/docs/plans/M5b/files/crates/brokerd/tests/invalid_detail.rs new file mode 100644 index 0000000..a482cdc --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/invalid_detail.rs @@ -0,0 +1,258 @@ +//! An invalid call is denied with the parser's reason: `ArgsError::detail` makes it one short +//! line, `policy::decide` keeps it, and `broker.sock` sends it. The audit record is unchanged. +//! Do not edit. + +#[path = "support/build.rs"] +mod build; +#[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 brokerd::args::{ArgsError, MAX_DETAIL}; +use brokerd::policy::decide; +use build::{denied, fetch, grant, now, private, read, set, shell}; +use bxcore::{AuditEvent, DecisionRecord, DenyReason, Envelope, Message, Mode, ToolResponse}; +use client::{Serve, call}; +use rig::{Rig, grant_text}; +use runtime::Recording; + +#[test] +fn the_detail_is_the_error_text() { + assert_eq!(MAX_DETAIL, 300); + let cases = [ + ( + ArgsError::Shape("missing field `path`".to_string()), + "invalid argument shape: missing field `path`", + ), + ( + ArgsError::Path("notes/a.md".to_string()), + "invalid path: notes/a.md", + ), + ( + ArgsError::Url("http://example.com/".to_string()), + "invalid url: http://example.com/", + ), + ]; + for (error, want) in cases { + assert_eq!(error.detail(), want); + assert_eq!( + error.detail(), + error.to_string(), + "a short clean text is kept as it is" + ); + } +} + +/// Every control character, and U+2028 and U+2029 (the line and paragraph separators), becomes +/// one space each; other characters, spaces and non-ASCII letters included, are kept. +#[test] +fn the_detail_is_one_line() { + let error = ArgsError::Path( + "a\nb\r\nc\td\u{7}e\u{1b}[31mf\u{7f}g\u{85}h\u{2028}i\u{2029}j k \u{e9}\u{65e5}" + .to_string(), + ); + assert_eq!( + error.detail(), + "invalid path: a b c d e [31mf g h i j k \u{e9}\u{65e5}" + ); + let error = ArgsError::Shape("line one\nline two\n".to_string()); + assert_eq!(error.detail(), "invalid argument shape: line one line two "); +} + +/// At most `MAX_DETAIL` characters, counted as characters, not bytes: a cut never splits one. +#[test] +fn the_detail_is_cut_to_300_characters() { + // "invalid path: " is 14 characters. + let exact = ArgsError::Path("a".repeat(286)); + assert_eq!( + exact.detail(), + exact.to_string(), + "300 characters are kept whole" + ); + let over = ArgsError::Path("a".repeat(287)); + assert_eq!(over.detail(), format!("invalid path: {}", "a".repeat(286))); + + let wide = ArgsError::Path("\u{e9}".repeat(1000)); + let detail = wide.detail(); + assert_eq!(detail.chars().count(), 300); + assert_eq!(detail, format!("invalid path: {}", "\u{e9}".repeat(286))); + + // Newlines past the cut do not matter; newlines before it are spaces. + let long = ArgsError::Shape(format!("{}\n{}", "x".repeat(10), "y\n".repeat(500))); + let detail = long.detail(); + assert_eq!(detail.chars().count(), 300); + assert!(!detail.contains('\n'), "{detail:?}"); + assert!(detail.starts_with("invalid argument shape: xxxxxxxxxx y y")); +} + +/// Each kind of `ArgsError`, from the model's own argument text, reaches the denial as its detail, +/// with no grant and with a grant that would cover the call: arguments are checked first. (The +/// memory tools take no grant: their policy is built in.) +#[test] +fn each_kind_of_invalid_arguments_carries_its_detail() { + let covering = set(vec![ + grant("r", "read_file", Mode::Auto).paths(&["/n"]), + grant("s", "shell", Mode::Auto), + grant("f", "http_fetch", Mode::Auto).hosts(&["example.com"]), + ]); + let none = set(vec![]); + let cases = [ + ( + build::request( + "memory", + r#"{"action":"note","file":"notes.md","text":"x"}"#, + ), + "invalid argument shape: unknown field `file`, expected `text`", + ), + ( + build::request("read_file", r#"{"path":"/n/a","file":"x"}"#), + "invalid argument shape: unknown field `file`, expected `path` at line 1 column 21", + ), + ( + build::request("read_file", "not json"), + "invalid argument shape: expected ident at line 1 column 2", + ), + ( + build::request("read_file", "{}"), + "invalid argument shape: missing field `path` at line 1 column 2", + ), + ( + build::request("recall", r#"{"query":""}"#), + "invalid argument shape: query must have 1 to 500 characters", + ), + (read("n/a"), "invalid path: n/a"), + (shell(Some("a\\nb")), "invalid path: a b"), + ( + fetch("http://example.com/"), + "invalid url: http://example.com/", + ), + ]; + for grants in [&none, &covering] { + for (req, want) in &cases { + let text = req.arguments.clone(); + let denial = denied(decide(req.clone(), grants, private(), now())); + assert_eq!(denial.reason, DenyReason::InvalidArguments, "{text}"); + assert_eq!(denial.detail.as_deref(), Some(*want), "{text}"); + assert_eq!(denial.grant, None, "{text}"); + } + } +} + +#[test] +fn a_long_invalid_path_is_cut_in_the_denial() { + let path = format!("/{}", "a".repeat(5000)); + let denial = denied(decide(read(&path), &set(vec![]), private(), now())); + assert_eq!(denial.reason, DenyReason::InvalidArguments); + let detail = denial.detail.unwrap(); + assert_eq!(detail.chars().count(), 300); + assert_eq!(detail, format!("invalid path: /{}", "a".repeat(285))); +} + +/// Only an invalid-arguments denial has a detail. +#[test] +fn other_denials_have_no_detail() { + let grants = set(vec![ + grant("no", "read_file", Mode::Deny).paths(&["/n"]), + grant("old", "shell", Mode::Auto).expires("2026-09-01T00:00:00.000Z"), + ]); + let cases = [ + (build::request("echo", "not json"), DenyReason::NoGrant), + (fetch("https://example.com/"), DenyReason::NoGrant), + (read("/n/a"), DenyReason::DeniedByGrant), + (shell(None), DenyReason::GrantExpired), + ]; + for (req, reason) in cases { + let text = req.arguments.clone(); + let denial = denied(decide(req, &grants, private(), now())); + assert_eq!(denial.reason, reason, "{text}"); + assert_eq!(denial.detail, None, "{text}"); + } +} + +fn only(frames: &[Envelope]) -> &ToolResponse { + assert_eq!(frames.len(), 1, "{frames:?}"); + assert!(frames[0].r#final); + match &frames[0].msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +#[test] +fn broker_sock_sends_the_detail_and_the_audit_record_does_not_hold_it() { + let rig = Rig::new("broker-invalid-detail"); + rig.grant( + "notes", + &grant_text("read_file", "auto", "", "paths = [\"/n\"]"), + ); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let arguments = r#"{"path":"/n/a","file":"x"}"#; + let frames = call(&broker, rig::request("s1", 5, "read_file", arguments)); + let detail = + "invalid argument shape: unknown field `file`, expected `path` at line 1 column 21"; + assert_eq!( + only(&frames), + &ToolResponse::Denied { + reason: DenyReason::InvalidArguments, + detail: Some(detail.to_string()), + } + ); + assert_eq!(rt.count(), 0, "nothing ran"); + match rig.events().as_slice() { + [ + AuditEvent::Decision { + outcome, + arguments: recorded, + .. + }, + ] => { + assert_eq!( + *outcome, + DecisionRecord::Denied { + reason: DenyReason::InvalidArguments + } + ); + assert_eq!(recorded, arguments, "the record keeps the arguments"); + } + other => panic!("{other:?}"), + } + let audit = rig.cfg.audit_dir(); + let mut logs = 0; + for entry in std::fs::read_dir(&audit).unwrap() { + let path = entry.unwrap().path(); + if path.extension().is_none_or(|e| e != "jsonl") { + continue; + } + logs += 1; + let text = std::fs::read_to_string(&path).unwrap(); + assert!(!text.contains("unknown field"), "{text}"); + assert!(!text.contains("\"detail\""), "{text}"); + } + assert_eq!(logs, 1, "one audit log file"); +} + +#[test] +fn a_call_no_grant_allows_is_denied_on_broker_sock_without_a_detail() { + let rig = Rig::new("broker-no-detail"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let frames = call( + &broker, + rig::request("s1", 6, "read_file", r#"{"path":"/n/a"}"#), + ); + assert_eq!( + only(&frames), + &ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + } + ); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/ledger.rs b/docs/plans/M5b/files/crates/brokerd/tests/ledger.rs new file mode 100644 index 0000000..b9b8e59 --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/ledger.rs @@ -0,0 +1,425 @@ +//! The ledger's first and third steps: decide and record; raise the state and record the result. +//! And what happens after a failed append or a panic. Do not edit. + +#[path = "support/build.rs"] +mod build; +#[path = "support/rig.rs"] +mod rig; +#[path = "support/sink.rs"] +mod sink; +#[path = "support/tmp.rs"] +mod tmp; + +use std::os::unix::fs::PermissionsExt; +use std::panic::AssertUnwindSafe; + +use brokerd::ledger::{Call, Decided, Grants, NOT_RECORDED, POISONED, STOPPED}; +use brokerd::policy::{Denial, Label, SessionState}; +use build::{grant, now, read, set}; +use bxcore::{ + AuditEvent, CallId, DataClass, DecisionRecord, DenyReason, GrantProblem, Mode, ResultStatus, + SessionId, ToolResponse, +}; +use rig::Rig; + +fn one(grants: Vec) -> Grants { + Ok(set(grants)) +} + +fn notes(mode: Mode) -> Grants { + one(vec![grant("n", "read_file", mode).paths(&["/n"])]) +} + +fn call(label: Label) -> Call { + Call { + session: SessionId::new("s1").unwrap(), + call: CallId(1), + decision: 0, + label, + } +} + +fn secret() -> Label { + Label { + class: DataClass::Secret, + untrusted: true, + } +} + +fn result(content: &str, label: Label) -> ToolResponse { + ToolResponse::Result { + content: content.to_string(), + class: label.class, + untrusted: label.untrusted, + truncated: false, + } +} + +#[test] +fn an_allowed_call_is_recorded_in_full() { + let rig = Rig::new("ledger-allowed"); + let ledger = rig.ledger(); + let request = read("/n/a"); + let arguments = request.arguments.clone(); + match ledger.decide(request, ¬es(Mode::Auto), now()) { + Decided::Allowed { decision, seq } => { + assert_eq!(seq, 0); + assert_eq!(decision.grant(), "n"); + } + other => panic!("{other:?}"), + } + let records = rig.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].time, now()); + assert_eq!( + records[0].event, + AuditEvent::Decision { + session: SessionId::new("s1").unwrap(), + call: CallId(1), + tool: "read_file".to_string(), + arguments, + outcome: DecisionRecord::Allowed {}, + grant: Some("n".to_string()), + grant_sha256: Some(bxcore::sha256(b"n").unwrap()), + taint: DataClass::Private, + untrusted: false, + } + ); +} + +#[test] +fn an_ask_is_recorded_as_ask_with_the_state_it_was_decided_at() { + let rig = Rig::new("ledger-ask"); + let ledger = rig.ledger(); + match ledger.decide(read("/n/a"), ¬es(Mode::Ask), now()) { + Decided::Ask { ask, seq, state } => { + assert_eq!((seq, ask.grant()), (0, "n")); + assert_eq!(state, SessionState::default()); + } + other => panic!("{other:?}"), + } + match &rig.events()[0] { + AuditEvent::Decision { outcome, grant, .. } => { + assert_eq!(*outcome, DecisionRecord::Ask {}); + assert_eq!(grant.as_deref(), Some("n")); + } + other => panic!("{other:?}"), + } +} + +#[test] +fn a_denial_names_a_grant_only_when_a_grant_denied() { + let rig = Rig::new("ledger-denied"); + let ledger = rig.ledger(); + let none = ledger.decide(read("/n/a"), &one(Vec::new()), now()); + assert!( + matches!( + none, + Decided::Denied(Denial { + reason: DenyReason::NoGrant, + .. + }) + ), + "{none:?}" + ); + let deny = one(vec![grant("d", "read_file", Mode::Deny).paths(&["/n"])]); + let by = ledger.decide(read("/n/a"), &deny, now()); + assert!( + matches!( + by, + Decided::Denied(Denial { + reason: DenyReason::DeniedByGrant, + .. + }) + ), + "{by:?}" + ); + let events = rig.events(); + let named: Vec<(DecisionRecord, Option)> = events + .iter() + .map(|e| match e { + AuditEvent::Decision { outcome, grant, .. } => (outcome.clone(), grant.clone()), + other => panic!("{other:?}"), + }) + .collect(); + assert_eq!( + named, + [ + ( + DecisionRecord::Denied { + reason: DenyReason::NoGrant + }, + None + ), + ( + DecisionRecord::Denied { + reason: DenyReason::DeniedByGrant + }, + Some("d".to_string()) + ), + ] + ); +} + +#[test] +fn invalid_grants_come_first_then_an_unreadable_state() { + let rig = Rig::new("ledger-order"); + let ledger = rig.ledger(); + std::fs::create_dir_all(rig.cfg.state_dir()).unwrap(); + std::fs::write(rig.state_file("s1"), "not json").unwrap(); + let invalid: Grants = Err(vec![GrantProblem { + file: "x.toml".to_string(), + line: None, + problem: "bad".to_string(), + }]); + let first = ledger.decide(read("/n/a"), &invalid, now()); + assert!( + matches!( + first, + Decided::Denied(Denial { + reason: DenyReason::GrantsInvalid, + .. + }) + ), + "{first:?}" + ); + let second = ledger.decide(read("/n/a"), ¬es(Mode::Auto), now()); + assert!( + matches!( + second, + Decided::Denied(Denial { + reason: DenyReason::StateUnreadable, + .. + }) + ), + "{second:?}" + ); + // A state brokerd cannot read is recorded as the most sensitive one. + for event in rig.events() { + match event { + AuditEvent::Decision { + taint, untrusted, .. + } => assert_eq!((taint, untrusted), (DataClass::Secret, true)), + other => panic!("{other:?}"), + } + } + let damaged = rig.lines.with("see docs/runbook.md#broker-state-damaged"); + assert!(!damaged.is_empty(), "{:?}", rig.lines.all()); +} + +#[test] +fn a_failed_append_denies_this_call_and_every_later_one() { + let rig = Rig::new("ledger-stop"); + let ledger = rig.ledger(); + rig.switch.fail(true); + let first = ledger.decide(read("/n/a"), ¬es(Mode::Auto), now()); + assert!( + matches!( + first, + Decided::Denied(Denial { + reason: DenyReason::AuditUnavailable, + .. + }) + ), + "{first:?}" + ); + assert_eq!(rig.switch.attempts(), 1); + // The sink would work again; the ledger does not try it. + rig.switch.fail(false); + let later = ledger.decide(read("/n/a"), ¬es(Mode::Auto), now()); + assert!( + matches!( + later, + Decided::Denied(Denial { + reason: DenyReason::AuditUnavailable, + .. + }) + ), + "{later:?}" + ); + let finished = ledger.finish(&call(secret()), result("x", secret()), now()); + assert_eq!( + finished, + ToolResponse::Failed { + message: NOT_RECORDED.to_string() + } + ); + assert_eq!(rig.switch.attempts(), 1, "nothing more was written"); + assert!(rig.records().is_empty()); + assert!(!rig.state_file("s1").exists(), "the state was not raised"); + assert!(!rig.lines.with(STOPPED).is_empty()); + assert!( + rig.lines + .all() + .iter() + .all(|l| !l.contains("runbook") || l.ends_with("see docs/runbook.md#audit-unavailable")), + "{:?}", + rig.lines.all() + ); +} + +#[test] +fn a_panic_while_holding_the_ledger_denies_every_later_call() { + let rig = Rig::new("ledger-poison"); + let ledger = rig.ledger(); + rig.switch.panic_next(); + let panicked = std::panic::catch_unwind(AssertUnwindSafe(|| { + ledger.decide(read("/n/a"), ¬es(Mode::Auto), now()) + })); + assert!(panicked.is_err()); + let later = ledger.decide(read("/n/a"), ¬es(Mode::Auto), now()); + assert!( + matches!( + later, + Decided::Denied(Denial { + reason: DenyReason::AuditUnavailable, + .. + }) + ), + "{later:?}" + ); + assert_eq!(rig.switch.attempts(), 1); + assert!( + !rig.lines.with(POISONED).is_empty(), + "{:?}", + rig.lines.all() + ); + assert!(POISONED.ends_with("see docs/runbook.md#audit-unavailable")); +} + +#[test] +fn a_result_raises_the_state_then_is_recorded() { + let rig = Rig::new("ledger-result"); + let ledger = rig.ledger(); + let answer = ledger.finish(&call(secret()), result("key", secret()), now()); + assert_eq!(answer, result("key", secret()), "passed on unchanged"); + let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap(); + assert_eq!( + state, + SessionState { + taint: DataClass::Secret, + untrusted: true + } + ); + assert_eq!( + rig.events(), + [AuditEvent::Result { + session: SessionId::new("s1").unwrap(), + call: CallId(1), + decision: 0, + status: ResultStatus::Result, + class: DataClass::Secret, + untrusted: true, + truncated: false, + bytes: 3, + sha256: bxcore::sha256(b"key").unwrap(), + taint_after: DataClass::Secret, + }] + ); +} + +#[test] +fn taint_never_goes_down() { + let rig = Rig::new("ledger-down"); + let ledger = rig.ledger(); + let public = Label { + class: DataClass::Public, + untrusted: false, + }; + ledger.finish(&call(secret()), result("a", secret()), now()); + ledger.finish(&call(public), result("b", public), now()); + let after: Vec = rig + .events() + .iter() + .map(|e| match e { + AuditEvent::Result { taint_after, .. } => *taint_after, + other => panic!("{other:?}"), + }) + .collect(); + assert_eq!(after, [DataClass::Secret, DataClass::Secret]); + let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap(); + assert_eq!((state.taint, state.untrusted), (DataClass::Secret, true)); +} + +#[test] +fn a_failure_changes_no_state_and_is_recorded_by_its_message() { + let rig = Rig::new("ledger-failed"); + let ledger = rig.ledger(); + let failed = ToolResponse::Failed { + message: "the tool timed out".to_string(), + }; + assert_eq!( + ledger.finish(&call(secret()), failed.clone(), now()), + failed + ); + assert!(!rig.state_file("s1").exists()); + match &rig.events()[0] { + AuditEvent::Result { + status, + class, + bytes, + sha256, + taint_after, + .. + } => { + assert_eq!(*status, ResultStatus::Failed); + assert_eq!( + *class, + DataClass::Secret, + "the label the result would have had" + ); + assert_eq!(*bytes, 18); + assert_eq!(*sha256, bxcore::sha256(b"the tool timed out").unwrap()); + assert_eq!(*taint_after, DataClass::Private, "the taint before"); + } + other => panic!("{other:?}"), + } +} + +#[test] +fn a_state_that_cannot_be_written_withholds_the_content() { + if tmp::running_as_root("a_state_that_cannot_be_written_withholds_the_content") { + return; + } + let rig = Rig::new("ledger-ro"); + let ledger = rig.ledger(); + let dir = rig.cfg.state_dir(); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let answer = ledger.finish(&call(secret()), result("key", secret()), now()); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!( + answer, + ToolResponse::Failed { + message: NOT_RECORDED.to_string() + } + ); + assert!( + rig.records().is_empty(), + "no Result for a state not on disk" + ); + assert!( + !rig.lines + .with("see docs/runbook.md#broker-state-damaged") + .is_empty() + ); +} + +#[test] +fn a_result_record_that_cannot_be_written_withholds_the_content() { + let rig = Rig::new("ledger-norecord"); + let ledger = rig.ledger(); + rig.switch.fail(true); + let answer = ledger.finish(&call(secret()), result("key", secret()), now()); + assert_eq!( + answer, + ToolResponse::Failed { + message: NOT_RECORDED.to_string() + } + ); + assert!( + !rig.lines + .with("see docs/runbook.md#audit-unavailable") + .is_empty() + ); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/serve.rs b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs new file mode 100644 index 0000000..bcdffac --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs @@ -0,0 +1,411 @@ +//! `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. + 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 { + 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 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, + 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/M5b/files/crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json new file mode 100644 index 0000000..7bffeb0 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json @@ -0,0 +1 @@ +{"v":1,"id":9,"final":true,"msg":{"kind":"tool_response","body":{"status":"denied","reason":"invalid_arguments","detail":"invalid argument shape: unknown field `file`, expected `text`"}}} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/strict.rs b/docs/plans/M5b/files/crates/bxcore/tests/strict.rs new file mode 100644 index 0000000..c71c325 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/strict.rs @@ -0,0 +1,151 @@ +//! Every JSON object in every fixture must reject an unknown key. Do not edit. +//! +//! The other test files check unknown fields in a few hand-picked places. This one checks all of +//! them: it walks each fixture, adds one unknown key to one object at a time, at every depth, and +//! requires that the result no longer decodes. + +use bxcore::{AuditRecord, Envelope, Grant, LogRecord}; +use serde::de::DeserializeOwned; +use serde_json::Value; + +/// Every copy of `value` that has exactly one extra key in exactly one object. +fn with_one_unknown_key(value: &Value) -> Vec { + let mut out = Vec::new(); + match value { + Value::Object(map) => { + let mut extended = map.clone(); + extended.insert("zz_unknown".to_string(), Value::Bool(true)); + out.push(Value::Object(extended)); + for (key, child) in map { + for changed in with_one_unknown_key(child) { + let mut copy = map.clone(); + copy.insert(key.clone(), changed); + out.push(Value::Object(copy)); + } + } + } + Value::Array(items) => { + for (i, child) in items.iter().enumerate() { + for changed in with_one_unknown_key(child) { + let mut copy = items.clone(); + copy[i] = changed; + out.push(Value::Array(copy)); + } + } + } + _ => {} + } + out +} + +/// Returns how many variations were tried, so callers can check the walk reached nested objects. +fn check(what: &str, text: &str) -> usize { + let value: Value = serde_json::from_str(text).unwrap_or_else(|e| panic!("{what}: {e}")); + assert!( + serde_json::from_value::(value.clone()).is_ok(), + "{what}: fixture must decode" + ); + let variations = with_one_unknown_key(&value); + for changed in &variations { + assert!( + serde_json::from_value::(changed.clone()).is_err(), + "{what}: accepted an unknown key: {changed}" + ); + } + variations.len() +} + +fn fixture(path: &str) -> String { + let full = format!("{}/tests/fixtures/{path}", env!("CARGO_MANIFEST_DIR")); + std::fs::read_to_string(&full).unwrap_or_else(|e| panic!("{full}: {e}")) +} + +#[test] +fn envelopes_reject_unknown_keys_at_every_depth() { + for name in [ + "tool_request.json", + "tool_response_pending.json", + "tool_response_result.json", + "tool_response_failed.json", + "tool_response_denied.json", + "tool_response_denied_detail.json", + "error.json", + "turn.json", + "turn_event_tool_result.json", + "turn_event_content.json", + "turn_event_retrying.json", + "turn_done.json", + "error_session_full.json", + "approvals.json", + "approval_list.json", + "approval_list_empty.json", + "approve.json", + "approve_result_allowed.json", + "approve_result_denied.json", + "refuse.json", + "refuse_no_reason.json", + "ok.json", + "check_grants.json", + "grants_report.json", + "grants_report_ok.json", + "error_forbidden.json", + "error_no_such_approval.json", + "turn_event_approval_pending.json", + "turn_event_tool_denied.json", + ] { + // Envelope, msg and body: three objects. Some bodies hold more: turn_done a usage + // object, approval_list one item, approve_result an outcome, grants_report two problems. + let want = match name { + "turn_done.json" | "approval_list.json" => 4, + "approve_result_allowed.json" | "approve_result_denied.json" => 4, + "grants_report.json" => 5, + _ => 3, + }; + assert_eq!( + check::(name, &fixture(&format!("wire/{name}"))), + want, + "{name}" + ); + } +} + +#[test] +fn audit_records_reject_unknown_keys_at_every_depth() { + for (i, line) in fixture("records/audit.jsonl").lines().enumerate() { + // The record and its event: two objects. Decision and approval events also hold an + // outcome object. + let want = if line.contains("\"outcome\"") { 3 } else { 2 }; + assert_eq!( + check::(&format!("audit.jsonl:{}", i + 1), line), + want + ); + } +} + +#[test] +fn log_records_reject_unknown_keys_at_every_depth() { + let mut tried = 0; + for (i, line) in fixture("records/session.jsonl").lines().enumerate() { + tried += check::(&format!("session.jsonl:{}", i + 1), line); + } + // Seven records, plus the one tool call inside the first assistant record. + assert_eq!(tried, 8); +} + +#[test] +fn usage_log_records_reject_unknown_keys_at_every_depth() { + let mut tried = 0; + for (i, line) in fixture("records/session_usage.jsonl").lines().enumerate() { + tried += check::(&format!("session_usage.jsonl:{}", i + 1), line); + } + // Seven records, plus the one tool call inside the first assistant record. + assert_eq!(tried, 8); +} + +#[test] +fn grants_reject_unknown_keys_at_every_depth() { + let grant: Grant = toml::from_str(&fixture("grant/full.toml")).unwrap(); + let text = serde_json::to_string(&grant).unwrap(); + // The grant and its constraints: two objects. + assert_eq!(check::("full.toml as JSON", &text), 2); +} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/wire.rs b/docs/plans/M5b/files/crates/bxcore/tests/wire.rs new file mode 100644 index 0000000..9d2e977 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/wire.rs @@ -0,0 +1,219 @@ +//! Tests for IPC messages against byte-exact fixtures. Do not edit these or the fixtures. + +use bxcore::{ + CallId, DataClass, DenyReason, Envelope, ErrorCode, Message, SessionId, Timestamp, ToolRequest, + ToolResponse, WireError, +}; + +fn fixture(name: &str) -> String { + let path = format!("{}/tests/fixtures/wire/{name}", env!("CARGO_MANIFEST_DIR")); + let text = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("{path}: {e}")); + text.trim_end_matches('\n').to_string() +} + +/// The fixture must decode to `want`, and `want` must encode to exactly the fixture's bytes. +fn check(name: &str, want: Envelope) { + let text = fixture(name); + let got: Envelope = serde_json::from_str(&text).unwrap_or_else(|e| panic!("{name}: {e}")); + assert_eq!(got, want, "{name}: decoded value"); + assert_eq!( + serde_json::to_string(&want).unwrap(), + text, + "{name}: encoded bytes" + ); +} + +fn response(id: u64, r#final: bool, body: ToolResponse) -> Envelope { + Envelope { + v: 1, + id, + r#final, + msg: Message::ToolResponse(body), + } +} + +#[test] +fn tool_request() { + let body = ToolRequest { + session: SessionId::new("mm-thread-42").unwrap(), + call: CallId(3), + tool: "read_file".to_string(), + arguments: r#"{"path":"/etc/hosts"}"#.to_string(), + }; + check( + "tool_request.json", + Envelope { + v: 1, + id: 7, + r#final: true, + msg: Message::ToolRequest(body), + }, + ); +} + +#[test] +fn tool_response_pending() { + let body = ToolResponse::PendingApproval { + approval: 41, + expires: Timestamp::parse("2026-09-17T08:35:00.000Z").unwrap(), + }; + check("tool_response_pending.json", response(7, false, body)); +} + +#[test] +fn an_approval_id_is_a_number_not_a_string() { + let good = fixture("tool_response_pending.json"); + let bad = good.replacen("\"approval\":41", "\"approval\":\"41\"", 1); + assert_ne!(good, bad); + assert!(serde_json::from_str::(&good).is_ok()); + assert!(serde_json::from_str::(&bad).is_err()); + let negative = good.replacen("\"approval\":41", "\"approval\":-1", 1); + assert!(serde_json::from_str::(&negative).is_err()); +} + +#[test] +fn tool_response_result() { + let body = ToolResponse::Result { + content: "127.0.0.1 localhost\n".to_string(), + class: DataClass::Private, + untrusted: true, + truncated: false, + }; + check("tool_response_result.json", response(7, true, body)); +} + +#[test] +fn tool_response_failed() { + let body = ToolResponse::Failed { + message: "exit status 2".to_string(), + }; + check("tool_response_failed.json", response(8, true, body)); +} + +#[test] +fn tool_response_denied() { + let body = ToolResponse::Denied { + reason: DenyReason::TaintTooHigh, + detail: None, + }; + check("tool_response_denied.json", response(9, true, body)); +} + +/// A detail is written only when there is one, so the fixture above has no `detail` key. +#[test] +fn tool_response_denied_with_a_detail() { + let body = ToolResponse::Denied { + reason: DenyReason::InvalidArguments, + detail: Some("invalid argument shape: unknown field `file`, expected `text`".to_string()), + }; + check("tool_response_denied_detail.json", response(9, true, body)); +} + +#[test] +fn a_detail_must_be_a_string() { + let good = fixture("tool_response_denied_detail.json"); + assert!(serde_json::from_str::(&good).is_ok()); + let at = good.find("\"detail\":").unwrap(); + for value in ["7", "[\"a\"]", "{}", "true"] { + let bad = format!("{}\"detail\":{value}}}}}}}", &good[..at]); + assert!( + serde_json::from_str::(&bad).is_err(), + "accepted {bad}" + ); + } +} + +#[test] +fn error_message() { + let body = WireError { + code: ErrorCode::BadVersion, + detail: "expected 1".to_string(), + }; + check( + "error.json", + Envelope { + v: 1, + id: 0, + r#final: true, + msg: Message::Error(body), + }, + ); +} + +#[test] +fn deny_reasons_and_error_codes_are_snake_case() { + let reasons = [ + (DenyReason::NoGrant, "no_grant"), + (DenyReason::GrantExpired, "grant_expired"), + (DenyReason::TaintTooHigh, "taint_too_high"), + (DenyReason::DeniedByGrant, "denied_by_grant"), + (DenyReason::ApprovalRefused, "approval_refused"), + (DenyReason::ApprovalExpired, "approval_expired"), + (DenyReason::GrantsInvalid, "grants_invalid"), + (DenyReason::AuditUnavailable, "audit_unavailable"), + (DenyReason::InvalidArguments, "invalid_arguments"), + (DenyReason::StateUnreadable, "state_unreadable"), + ]; + for (value, text) in reasons { + assert_eq!( + serde_json::to_string(&value).unwrap(), + format!("\"{text}\"") + ); + } + let codes = [ + (ErrorCode::BadFrame, "bad_frame"), + (ErrorCode::BadVersion, "bad_version"), + (ErrorCode::BadMessage, "bad_message"), + (ErrorCode::Internal, "internal"), + (ErrorCode::Forbidden, "forbidden"), + (ErrorCode::NoSuchApproval, "no_such_approval"), + ]; + for (value, text) in codes { + assert_eq!( + serde_json::to_string(&value).unwrap(), + format!("\"{text}\"") + ); + } +} + +#[test] +fn unknown_and_missing_fields_are_rejected() { + let good = fixture("tool_request.json"); + assert!(serde_json::from_str::(&good).is_ok()); + let bad = [ + // extra field in the envelope + good.replacen("{\"v\":1,", "{\"v\":1,\"extra\":0,", 1), + // extra field beside kind and body + good.replacen( + "\"kind\":\"tool_request\",", + "\"kind\":\"tool_request\",\"x\":1,", + 1, + ), + // extra field in the body + good.replacen("\"call\":3,", "\"call\":3,\"priority\":9,", 1), + // missing field in the body + good.replacen("\"call\":3,", "", 1), + // missing `final` + good.replacen("\"final\":true,", "", 1), + // unknown kind + good.replacen("tool_request", "tool_demand", 1), + // invalid session id inside a message + good.replacen("mm-thread-42", "../../etc", 1), + ]; + for text in bad { + assert!( + serde_json::from_str::(&text).is_err(), + "accepted {text}" + ); + } +} + +#[test] +fn unknown_field_in_a_response_variant_is_rejected() { + let good = fixture("tool_response_denied.json"); + let bad = good.replacen("\"reason\":", "\"note\":\"x\",\"reason\":", 1); + assert!(serde_json::from_str::(&good).is_ok()); + assert!(serde_json::from_str::(&bad).is_err()); + let bad_status = good.replacen("denied", "refused", 1); + assert!(serde_json::from_str::(&bad_status).is_err()); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/broker_port.rs b/docs/plans/M5b/files/crates/loopd/tests/broker_port.rs new file mode 100644 index 0000000..b1981bf --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/broker_port.rs @@ -0,0 +1,321 @@ +//! Tests for `BrokerPort` against a fake `brokerd` on a temporary socket: answers, approvals and +//! timeouts. Frames that break the protocol are in `broker_port_bad.rs`. Do not edit. + +#[path = "support/broker.rs"] +mod fake; + +use std::io::Write; +use std::os::unix::net::UnixListener; +use std::thread; +use std::time::Duration; + +use bxcore::{ + DenyReason, Message, PROTOCOL_VERSION, Timestamp, ToolResponse, read_frame, write_frame, +}; +use fake::{assert_unavailable, broker, call, frame, in_ms, request, result, send, socket_path}; +use loopd::broker_port::{BrokerPort, NoBroker, POINTER, TOO_LARGE}; +use loopd::tools::{Pending, ToolPort}; + +#[test] +fn a_result_comes_back_as_it_was_sent() { + let (socket, broker) = broker(|stream, request| { + send(stream, request.id, true, result("hello\n")); + }); + let got = call(socket, 2_000, &request()); + assert_eq!(got.response, result("hello\n")); + assert!(got.pending.is_empty()); + assert!(got.lines.is_empty(), "nothing to report: {:?}", got.lines); + + // What the broker received: one final frame holding exactly the request. + let sent = broker.join().unwrap(); + assert_eq!(sent.v, PROTOCOL_VERSION); + assert!(sent.r#final, "a request is a single final frame"); + assert_eq!(sent.msg, Message::ToolRequest(request())); +} + +#[test] +fn every_denial_and_a_failure_come_back_as_they_were_sent() { + let mut answers: Vec = [ + DenyReason::NoGrant, + DenyReason::GrantExpired, + DenyReason::TaintTooHigh, + DenyReason::DeniedByGrant, + DenyReason::ApprovalRefused, + DenyReason::ApprovalExpired, + DenyReason::GrantsInvalid, + DenyReason::AuditUnavailable, + DenyReason::InvalidArguments, + DenyReason::StateUnreadable, + ] + .into_iter() + .map(|reason| ToolResponse::Denied { + reason, + detail: None, + }) + .collect(); + answers.push(ToolResponse::Denied { + reason: DenyReason::InvalidArguments, + detail: Some("invalid path: n/a".to_string()), + }); + answers.push(ToolResponse::Failed { + message: "the runner arrives in M3b".to_string(), + }); + for answer in answers { + let reply = answer.clone(); + let (socket, broker) = broker(move |stream, request| { + send(stream, request.id, true, reply); + }); + let got = call(socket, 2_000, &request()); + assert_eq!(got.response, answer); + assert!( + got.lines.is_empty(), + "a denial is not an outage: {:?}", + got.lines + ); + broker.join().unwrap(); + } +} + +#[test] +fn a_pending_frame_reaches_the_callback_and_the_final_frame_is_the_answer() { + let expires = in_ms(60_000); + let (socket, broker) = broker(move |stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 41, + expires, + }; + send(stream, request.id, false, pending); + thread::sleep(Duration::from_millis(150)); + send(stream, request.id, true, result("approved")); + }); + let got = call(socket, 2_000, &request()); + assert_eq!(got.response, result("approved")); + assert_eq!( + got.pending, + [Pending { + approval: 41, + expires + }], + "called once, with the frame's values" + ); + assert!(got.lines.is_empty(), "{:?}", got.lines); + broker.join().unwrap(); +} + +#[test] +fn an_answer_after_expires_but_within_the_timeout_of_it_is_delivered() { + // expires in 100 ms, timeout 1,500 ms: the port waits until 1,600 ms. The answer comes at + // about 400 ms, well after `expires`: an approval given at the last moment still gets the + // whole timeout to run. + let (socket, broker) = broker(|stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 1, + expires: in_ms(100), + }; + send(stream, request.id, false, pending); + thread::sleep(Duration::from_millis(400)); + send(stream, request.id, true, result("late but good")); + }); + let got = call(socket, 1_500, &request()); + assert_eq!(got.response, result("late but good")); + assert!(got.lines.is_empty(), "{:?}", got.lines); + broker.join().unwrap(); +} + +#[test] +fn an_expiry_that_has_already_passed_still_leaves_the_timeout() { + let (socket, broker) = broker(|stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 1, + expires: Timestamp::parse("2020-01-01T00:00:00.000Z").unwrap(), + }; + send(stream, request.id, false, pending); + thread::sleep(Duration::from_millis(200)); + send(stream, request.id, true, result("fine")); + }); + let got = call(socket, 1_500, &request()); + assert_eq!(got.response, result("fine")); + broker.join().unwrap(); +} + +#[test] +fn no_socket_is_unavailable() { + let socket = socket_path(); + let _ = std::fs::remove_file(&socket); + let got = call(socket.to_path_buf(), 2_000, &request()); + assert_unavailable(&got, "no socket file"); + assert!( + got.lines[0].contains(&socket.display().to_string()), + "the line names the socket: {}", + got.lines[0] + ); + assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took); +} + +#[test] +fn a_broker_that_closes_without_answering_is_unavailable() { + let (socket, broker) = broker(|_, _| {}); + let got = call(socket, 2_000, &request()); + assert_unavailable(&got, "closed before any frame"); + assert!( + got.took < Duration::from_millis(1_500), + "a close is seen at once: {:?}", + got.took + ); + broker.join().unwrap(); +} + +#[test] +fn a_broker_that_closes_while_pending_is_unavailable() { + let (socket, broker) = broker(|stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 3, + expires: in_ms(60_000), + }; + send(stream, request.id, false, pending); + // brokerd was restarted: the connection just ends. + }); + let got = call(socket, 2_000, &request()); + assert_unavailable(&got, "closed while pending"); + assert_eq!(got.pending.len(), 1, "the pending frame was reported first"); + assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took); + broker.join().unwrap(); +} + +#[test] +fn a_broker_that_never_answers_is_unavailable_after_the_timeout() { + let (socket, broker) = broker(|_, _| thread::sleep(Duration::from_millis(1_200))); + let got = call(socket, 300, &request()); + assert_unavailable(&got, "silence"); + assert!( + got.took >= Duration::from_millis(250), + "gave up early: {:?}", + got.took + ); + assert!( + got.took < Duration::from_millis(1_100), + "gave up late: {:?}", + got.took + ); + broker.join().unwrap(); +} + +#[test] +fn a_pending_call_nobody_answers_ends_at_expires_plus_the_timeout() { + let (socket, broker) = broker(|stream, request| { + let pending = ToolResponse::PendingApproval { + approval: 3, + expires: in_ms(300), + }; + send(stream, request.id, false, pending); + thread::sleep(Duration::from_millis(1_800)); + }); + let got = call(socket, 300, &request()); + assert_unavailable(&got, "silence while pending"); + assert!( + got.took >= Duration::from_millis(550), + "it must wait for expires (300) plus the timeout (300): {:?}", + got.took + ); + assert!( + got.took < Duration::from_millis(1_700), + "gave up late: {:?}", + got.took + ); + broker.join().unwrap(); +} + +#[test] +fn the_timeout_is_a_deadline_for_the_frame_not_for_each_read() { + // The four header bytes arrive 250 ms apart, the last at 1,000 ms. A port that only sets a + // 600 ms read timeout sees every single read succeed and returns the result; a port with a + // deadline gives up at 600 ms. The answer tells them apart, so no timing is asserted. + let (socket, broker) = broker(|stream, request| { + let mut bytes = Vec::new(); + write_frame( + &mut bytes, + &frame(request.id, true, Message::ToolResponse(result("slow"))), + ) + .unwrap(); + for byte in bytes.iter().take(4) { + thread::sleep(Duration::from_millis(250)); + if stream.write_all(&[*byte]).is_err() { + return; + } + } + let _ = stream.write_all(&bytes[4..]); + }); + let got = call(socket, 600, &request()); + assert_unavailable(&got, "a trickled frame"); + broker.join().unwrap(); +} + +#[test] +fn a_zero_timeout_fails_closed_and_does_not_panic() { + let (socket, _broker) = broker(|stream, request| { + send(stream, request.id, true, result("too late")); + }); + let got = call(socket, 0, &request()); + assert_unavailable(&got, "timeout_ms = 0"); +} + +#[test] +fn a_request_too_large_for_a_frame_is_its_own_failure() { + // Nothing is sent, so the fake broker sees a connection that closes or none at all. + let path = socket_path(); + let _ = std::fs::remove_file(&path); + let _listener = UnixListener::bind(&path).unwrap(); + let mut big = request(); + big.arguments = format!(r#"{{"path":"/{}"}}"#, "a".repeat(bxcore::MAX_FRAME)); + let got = call(path.to_path_buf(), 1_000, &big); + assert_eq!( + got.response, + ToolResponse::Failed { + message: TOO_LARGE.to_string() + } + ); + assert!( + got.lines.is_empty(), + "the broker is fine; this is not an outage: {:?}", + got.lines + ); +} + +#[test] +fn every_call_is_its_own_connection() { + let path = socket_path(); + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path).unwrap(); + let server = thread::spawn(move || { + for n in 0..3u64 { + let (mut stream, _) = listener.accept().unwrap(); + let request = read_frame(&mut stream).unwrap(); + send( + &mut stream, + request.id, + true, + result(&format!("answer {n}")), + ); + } + }); + let port = BrokerPort::new(path.to_path_buf(), Duration::from_millis(2_000)); + for n in 0..3 { + let got = port.call(&request(), &mut |_| {}); + assert_eq!(got, result(&format!("answer {n}"))); + } + server.join().unwrap(); +} + +#[test] +fn without_a_broker_every_call_fails_in_words_the_model_can_repeat() { + let mut seen = 0; + let got = NoBroker.call(&request(), &mut |_| seen += 1); + assert_eq!( + got, + ToolResponse::Failed { + message: "no tool broker is configured".to_string() + } + ); + assert_eq!(seen, 0); + assert_eq!(POINTER, "see docs/runbook.md#broker-unavailable"); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/tools.rs b/docs/plans/M5b/files/crates/loopd/tests/tools.rs new file mode 100644 index 0000000..2278fa2 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/tools.rs @@ -0,0 +1,335 @@ +//! Tests for the registry, dispatch, the denial sentences, the result cap and the fake tools. +//! Do not edit. + +mod support; + +use bxcore::{CallId, DenyReason, SessionId, ToolRequest, ToolResponse}; +use loopd::tools::{ + Dispatch, FakeTools, Pending, Registry, ToolPort, cap_result, denial_message, denial_text, + dispatch, +}; + +fn req(tool: &str, arguments: &str) -> ToolRequest { + ToolRequest { + session: SessionId::new("s").unwrap(), + call: CallId(1), + tool: tool.to_string(), + arguments: arguments.to_string(), + } +} + +#[test] +fn the_core_schemas_are_the_fixed_tools_array() { + let names: Vec = Registry::m2b() + .core_schemas() + .into_iter() + .map(|s| s.name) + .collect(); + assert_eq!(names, ["clock", "find_tool", "call_tool"]); + let find = &Registry::m2b().core_schemas()[1]; + assert_eq!(find.parameters["required"], serde_json::json!(["query"])); + let call = &Registry::m2b().core_schemas()[2]; + assert_eq!( + call.parameters["required"], + serde_json::json!(["name", "arguments"]) + ); +} + +#[test] +fn find_matches_name_or_description_case_insensitively() { + let r = Registry::m2b(); + let names = |q: &str| r.find(q).iter().map(|s| s.name.clone()).collect::>(); + assert_eq!(names("echo"), ["echo"]); + assert_eq!(names("ECHO"), ["echo"]); + assert_eq!(names("unchanged"), ["echo"], "matches the description too"); + assert_eq!(names("time"), ["clock"]); + assert!(names("weather").is_empty()); + assert!( + names("").is_empty(), + "an empty query matches nothing, not everything" + ); + assert!(names(" ").is_empty()); +} + +#[test] +fn find_tool_is_answered_locally() { + let r = Registry::m2b(); + match dispatch(&r, "find_tool", r#"{"query":"echo"}"#) { + Dispatch::Local(text) => { + assert!(text.starts_with("1 tool(s) match:\n"), "{text}"); + assert!(text.contains("\"name\":\"echo\""), "{text}"); + assert!( + text.contains("\"parameters\""), + "the schema is included: {text}" + ); + assert!(text.ends_with("Call it with call_tool."), "{text}"); + } + other => panic!("{other:?}"), + } + assert_eq!( + dispatch(&r, "find_tool", r#"{"query":"weather"}"#), + Dispatch::Local("No tool matches.".to_string()) + ); + assert!( + matches!(dispatch(&r, "find_tool", "not json"), Dispatch::Local(t) if t == "No tool matches.") + ); +} + +#[test] +fn call_tool_unwraps_a_known_non_core_tool_and_nothing_else() { + let r = Registry::m2b(); + let args = r#"{"name":"echo","arguments":{"text": "box"}}"#; + match dispatch(&r, "call_tool", args) { + Dispatch::Port { tool, arguments } => { + assert_eq!(tool, "echo"); + assert_eq!( + serde_json::from_str::(&arguments).unwrap(), + serde_json::json!({"text": "box"}) + ); + } + other => panic!("{other:?}"), + } + let cases = [ + ( + r#"{"name":"weather","arguments":{}}"#, + "No tool named \"weather\"", + ), + (r#"{"name":"clock","arguments":{}}"#, "core tool"), + (r#"{"arguments":{}}"#, "No tool named \"\""), + ("not json", "needs a JSON object"), + ]; + for (args, want) in cases { + match dispatch(&r, "call_tool", args) { + Dispatch::Local(text) => assert!(text.contains(want), "{args}: {text}"), + other => panic!("{args}: {other:?}, must never reach the port"), + } + } +} + +#[test] +fn any_other_tool_goes_to_the_port_as_it_is() { + let r = Registry::m2b(); + // Even one the registry does not know: the port (brokerd) decides, not loopd. + let want = Dispatch::Port { + tool: "read_file".to_string(), + arguments: r#"{"path":"/x"}"#.to_string(), + }; + assert_eq!(dispatch(&r, "read_file", r#"{"path":"/x"}"#), want); + let want = Dispatch::Port { + tool: "weather".to_string(), + arguments: "not json".to_string(), + }; + assert_eq!(dispatch(&r, "weather", "not json"), want); +} + +#[test] +fn the_clock_is_answered_locally_whatever_its_arguments() { + for registry in [Registry::m2b(), Registry::m3a(), Registry::new(vec![])] { + for arguments in ["{}", r#"{"zone":"UTC"}"#, "not json", ""] { + let before = bxcore::Timestamp::now(); + match dispatch(®istry, "clock", arguments) { + Dispatch::Local(text) => { + let time = bxcore::Timestamp::parse(&text) + .unwrap_or_else(|e| panic!("an RFC 3339 time, got {text:?}: {e:?}")); + assert!(time >= before, "{text}"); + assert!(text.ends_with('Z'), "UTC: {text}"); + } + other => panic!("{arguments:?}: {other:?}, the clock must not reach the port"), + } + } + } +} + +#[test] +fn the_m3a_registry_has_the_same_core_and_the_four_broker_tools() { + let r = Registry::m3a(); + assert_eq!( + r.core_schemas(), + Registry::m2b().core_schemas(), + "the tools array is part of the baseline: it must not change" + ); + let names = |q: &str| r.find(q).iter().map(|s| s.name.clone()).collect::>(); + assert_eq!(names("file"), ["read_file", "write_file"]); + assert_eq!(names("shell"), ["shell"]); + assert_eq!(names("fetch"), ["http_fetch"]); + assert_eq!(names("https"), ["http_fetch"]); + assert!(names("echo").is_empty(), "echo is a test tool only"); + + // The argument schemas are section 3's table: exactly these properties, all strings. + let table: [(&str, &[&str], &[&str]); 4] = [ + ("read_file", &["path"], &["path"]), + ("write_file", &["content", "path"], &["path", "content"]), + ("shell", &["command", "cwd"], &["command"]), + ("http_fetch", &["url"], &["url"]), + ]; + for (name, properties, required) in table { + let entry = r.get(name).unwrap_or_else(|| panic!("{name} is missing")); + assert!(!entry.core, "{name} is found with find_tool, not declared"); + let p = &entry.schema.parameters; + assert_eq!(p["type"], "object", "{name}"); + let mut got: Vec<&str> = p["properties"] + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + got.sort_unstable(); + assert_eq!(got, properties, "{name}: properties"); + for property in properties { + assert_eq!( + p["properties"][property]["type"], "string", + "{name}.{property}" + ); + assert!( + p["properties"][property]["description"].is_string(), + "{name}.{property} needs a description" + ); + } + assert_eq!( + p["required"], + serde_json::json!(required), + "{name}: required" + ); + } + + // call_tool lets the four through to the port and nothing else. + match dispatch( + &r, + "call_tool", + r#"{"name":"shell","arguments":{"command":"ls"}}"#, + ) { + Dispatch::Port { tool, arguments } => { + assert_eq!(tool, "shell"); + assert_eq!(arguments, r#"{"command":"ls"}"#); + } + other => panic!("{other:?}"), + } + assert!(matches!( + dispatch(&r, "call_tool", r#"{"name":"echo","arguments":{}}"#), + Dispatch::Local(t) if t.contains("No tool named \"echo\"") + )); +} + +#[test] +fn every_deny_reason_has_its_sentence() { + let table = [ + (DenyReason::NoGrant, "Denied: no grant allows this call."), + ( + DenyReason::GrantExpired, + "Denied: the grant for this call has expired.", + ), + ( + DenyReason::TaintTooHigh, + "Denied: this session has seen data too sensitive for this call.", + ), + ( + DenyReason::DeniedByGrant, + "Denied: a grant forbids this call.", + ), + ( + DenyReason::ApprovalRefused, + "Denied: the owner refused this call.", + ), + ( + DenyReason::ApprovalExpired, + "Denied: the approval request expired without an answer.", + ), + ( + DenyReason::InvalidArguments, + "Denied: the arguments are not valid for this tool.", + ), + ( + DenyReason::GrantsInvalid, + "Denied: the grant files have an error; the owner has been told.", + ), + ( + DenyReason::AuditUnavailable, + "Denied: the audit log cannot be written; the owner has been told.", + ), + ( + DenyReason::StateUnreadable, + "Denied: this session's broker state is damaged; the owner has been told.", + ), + ]; + for (reason, want) in table { + assert_eq!(denial_text(reason), want, "{reason:?}"); + assert_eq!( + denial_message(reason, None), + want, + "{reason:?}: without a detail, the sentence alone" + ); + } +} + +/// The broker's detail goes after the sentence, before its full stop. +#[test] +fn a_denial_with_a_detail_appends_it_to_the_sentence() { + assert_eq!( + denial_message( + DenyReason::InvalidArguments, + Some("invalid argument shape: unknown field `file`, expected `text`") + ), + "Denied: the arguments are not valid for this tool: invalid argument shape: unknown field `file`, expected `text`." + ); + assert_eq!( + denial_message(DenyReason::InvalidArguments, Some("invalid path: n/a")), + "Denied: the arguments are not valid for this tool: invalid path: n/a." + ); + assert_eq!( + denial_message(DenyReason::NoGrant, Some("x")), + "Denied: no grant allows this call: x.", + "any reason's detail is shown the same way" + ); +} + +#[test] +fn results_are_cut_on_a_character_boundary_and_marked() { + assert_eq!(cap_result("short", 100), ("short".to_string(), false)); + assert_eq!(cap_result("exactly", 7), ("exactly".to_string(), false)); + let (text, truncated) = cap_result("abcdefghij", 4); + assert_eq!(text, "abcd\n[truncated]"); + assert!(truncated); + // 3 ASCII bytes, then 2-byte characters: byte 4 is inside a character. + let (text, truncated) = cap_result("abc\u{e9}\u{e9}\u{e9}", 4); + assert_eq!(text, "abc\n[truncated]"); + assert!(truncated); + assert_eq!(cap_result("", 0), (String::new(), false)); + assert_eq!(cap_result("x", 0), ("\n[truncated]".to_string(), true)); +} + +#[test] +fn fake_tools_answer_echo_deny_the_rest_and_record_calls() { + let fake = FakeTools::new(); + let mut seen = 0; + let mut on_pending = |_: &Pending| seen += 1; + match fake.call(&req("echo", r#"{"text":"box"}"#), &mut on_pending) { + ToolResponse::Result { + content, + class, + untrusted, + truncated, + } => { + assert_eq!(content, "box"); + assert_eq!(class, bxcore::DataClass::Public); + assert!(!untrusted && !truncated); + } + other => panic!("{other:?}"), + } + assert!(matches!( + fake.call(&req("echo", r#"{"tex":"box"}"#), &mut on_pending), + ToolResponse::Failed { .. } + )); + for tool in ["weather", "read_file", "clock"] { + assert_eq!( + fake.call(&req(tool, "{}"), &mut on_pending), + ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + }, + "{tool}: the clock is loopd's own now, not the port's" + ); + } + assert_eq!(seen, 0, "the fake never asks for approval"); + assert_eq!(fake.calls().len(), 5); + assert_eq!(fake.calls()[0].tool, "echo"); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/turn_broker.rs b/docs/plans/M5b/files/crates/loopd/tests/turn_broker.rs new file mode 100644 index 0000000..66e9672 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/turn_broker.rs @@ -0,0 +1,269 @@ +//! Tests for what the turn loop does with the broker's answers: denials, pending approvals, and +//! a port that misbehaves. Do not edit. + +mod support; +#[path = "support/turn.rs"] +mod turn_support; + +use bxcore::{DataClass, DenyReason, LogRecord, Timestamp, ToolResponse, TurnEvent}; +use support::{Reply, ReturnsPendingPort, ok_result, pending}; +use turn_support::{setup, types}; + +const CHAT: &str = "/v1/chat/completions"; + +/// The tool events of a turn, in order, as short strings. +fn tool_events(events: &[TurnEvent]) -> Vec { + events + .iter() + .filter_map(|e| match e { + TurnEvent::ToolCallStarted { name } => Some(format!("started {name}")), + TurnEvent::ApprovalPending { approval, tool, .. } => { + Some(format!("pending {approval} {tool}")) + } + TurnEvent::ToolDenied { name, reason } => Some(format!("denied {name} {reason:?}")), + TurnEvent::ToolResult { name, .. } => Some(format!("result {name}")), + _ => None, + }) + .collect() +} + +#[test] +fn a_denial_is_a_fixed_sentence_for_the_model_and_an_event_for_the_owner() { + let s = setup(vec![ToolResponse::Denied { + reason: DenyReason::TaintTooHigh, + detail: None, + }]); + s.server.route( + CHAT, + vec![Reply::fixture("tool_call"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let (result, events) = s.turn(&mut session, "x"); + assert!( + result.is_ok(), + "the turn goes on after a denial: {result:?}" + ); + match &session.records()[4] { + LogRecord::ToolResult { + content, + class, + untrusted, + truncated, + .. + } => { + assert_eq!( + content, + "Denied: this session has seen data too sensitive for this call." + ); + assert_eq!( + (*class, *untrusted, *truncated), + (DataClass::Public, false, false) + ); + } + other => panic!("{other:?}"), + } + assert_eq!( + tool_events(&events), + [ + "started read_file", + "denied read_file TaintTooHigh", + "result read_file" + ], + "the denial comes before the result" + ); + // The model reads the sentence in the next request. + let m2 = s.server.requests_to(CHAT)[1].json(); + assert_eq!( + m2["messages"][3]["content"], + "Denied: this session has seen data too sensitive for this call." + ); +} + +/// The broker's detail reaches the model, in the log and in the next request; the owner's event +/// is the same as for any denial. +#[test] +fn an_invalid_call_tells_the_model_why() { + let s = setup(vec![ToolResponse::Denied { + reason: DenyReason::InvalidArguments, + detail: Some("invalid path: n/a".to_string()), + }]); + s.server.route( + CHAT, + vec![Reply::fixture("tool_call"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let (result, events) = s.turn(&mut session, "x"); + assert!(result.is_ok(), "{result:?}"); + let want = "Denied: the arguments are not valid for this tool: invalid path: n/a."; + match &session.records()[4] { + LogRecord::ToolResult { + content, + class, + untrusted, + truncated, + .. + } => { + assert_eq!(content, want); + assert_eq!( + (*class, *untrusted, *truncated), + (DataClass::Public, false, false) + ); + } + other => panic!("{other:?}"), + } + assert_eq!( + tool_events(&events), + [ + "started read_file", + "denied read_file InvalidArguments", + "result read_file" + ] + ); + let m2 = s.server.requests_to(CHAT)[1].json(); + assert_eq!(m2["messages"][3]["content"], want); +} + +#[test] +fn a_denied_call_tool_names_the_target_tool_in_the_denial() { + let s = setup(vec![ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + }]); + s.server.route( + CHAT, + vec![Reply::fixture("call_tool"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let (result, events) = s.turn(&mut session, "echo box"); + assert!(result.is_ok(), "{result:?}"); + assert_eq!( + tool_events(&events), + [ + "started call_tool", + "denied echo NoGrant", + "result call_tool" + ], + "the owner writes grants for `echo`, not for `call_tool`" + ); + assert!( + matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: no grant allows this call.") + ); +} + +#[test] +fn a_pending_approval_is_an_event_and_the_answer_after_it_is_the_result() { + let s = setup(vec![ + pending(41, "2026-09-18T12:15:00.000Z"), + ok_result("straylight\n"), + ]); + s.server.route( + CHAT, + vec![Reply::fixture("tool_call"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let (result, events) = s.turn(&mut session, "x"); + assert!(result.is_ok(), "{result:?}"); + assert_eq!( + tool_events(&events), + [ + "started read_file", + "pending 41 read_file", + "result read_file" + ] + ); + let expires = events.iter().find_map(|e| match e { + TurnEvent::ApprovalPending { expires, .. } => Some(*expires), + _ => None, + }); + assert_eq!( + expires, + Some(Timestamp::parse("2026-09-18T12:15:00.000Z").unwrap()) + ); + assert_eq!(s.port.calls().len(), 1, "one call, however long it waited"); + assert_eq!( + types(session.records()), + [ + "start", + "user", + "assistant", + "usage", + "tool_result", + "assistant", + "usage" + ], + "waiting writes nothing to the log" + ); + assert!( + matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "straylight\n") + ); +} + +#[test] +fn a_pending_approval_that_ends_in_a_refusal() { + let s = setup(vec![ + pending(7, "2026-09-18T12:15:00.000Z"), + ToolResponse::Denied { + reason: DenyReason::ApprovalRefused, + detail: None, + }, + ]); + s.server.route( + CHAT, + vec![Reply::fixture("tool_call"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let (result, events) = s.turn(&mut session, "x"); + assert!(result.is_ok(), "{result:?}"); + assert_eq!( + tool_events(&events), + [ + "started read_file", + "pending 7 read_file", + "denied read_file ApprovalRefused", + "result read_file" + ] + ); + assert!( + matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: the owner refused this call.") + ); +} + +#[test] +fn a_port_that_returns_a_pending_frame_as_its_answer_is_a_failure_not_a_decision() { + let s = setup(vec![]); + s.server.route( + CHAT, + vec![Reply::fixture("tool_call"), Reply::fixture("plain")], + ); + let mut session = s.session("a"); + let runtime = loopd::turn::Runtime { + cfg: &s.cfg, + client: &s.client, + port: &ReturnsPendingPort, + registry: &s.registry, + }; + let mut events = Vec::new(); + let result = + loopd::turn::run_turn(&mut session, &runtime, "x", &mut |e| events.push(e.clone())); + assert!(result.is_ok(), "{result:?}"); + match &session.records()[4] { + LogRecord::ToolResult { + content, + class, + untrusted, + .. + } => { + assert_eq!( + content, + "The tool failed: the tool broker gave no final answer" + ); + assert_eq!((*class, *untrusted), (DataClass::Public, false)); + } + other => panic!("{other:?}"), + } + assert_eq!( + tool_events(&events), + ["started read_file", "result read_file"], + "neither pending nor denied: the port said neither" + ); +}