diff --git a/crates/brokerd/src/broker.rs b/crates/brokerd/src/broker.rs new file mode 100644 index 0000000..7bfa837 --- /dev/null +++ b/crates/brokerd/src/broker.rs @@ -0,0 +1,313 @@ +//! Serving one `broker.sock` connection: decide, record, run if allowed, record the result, and +//! answer with one final frame carrying the request's id. An `ask` call first sends one pending +//! frame and waits for whoever takes the table entry to send the verdict. Every record goes +//! through the ledger. + +use std::io::Read; +use std::os::unix::net::UnixStream; +use std::sync::Mutex; +use std::sync::mpsc::RecvTimeoutError; +use std::time::Duration; + +use proto::{ + DenyReason, Envelope, ErrorCode, FrameError, Message, PROTOCOL_VERSION, PendingApproval, + Timestamp, ToolResponse, WireError, read_frame, write_frame, +}; + +use crate::approvals::{Table, Verdict}; +use crate::config::Config; +use crate::grants; +use crate::ledger::{Call, Decided, Grants, Ledger}; +use crate::policy::{Ask, SessionState}; +use crate::runner; +use crate::runner::Runtime; + +/// The failure message sent when a peer leaves at the last look, after the table entry is taken. +pub const GONE: &str = "the requester went away"; + +/// A line printer: one line per call, owned by the broker. +pub type Log = Box; + +/// The only role that holds authority: one runtime, the ledger, the approval table, and the grants +/// the owner can re-read. +pub struct Broker { + cfg: Config, + ledger: Ledger, + table: Table, + runtime: Box, + log: Log, + printed: Mutex>>, +} + +impl Broker { + pub fn new(cfg: Config, ledger: Ledger, runtime: Box, log: Log) -> Broker { + Broker { + cfg, + ledger, + table: Table::new(), + runtime, + log, + printed: Mutex::new(None), + } + } + + pub fn cfg(&self) -> &Config { + &self.cfg + } + + pub fn ledger(&self) -> &Ledger { + &self.ledger + } + + pub fn table(&self) -> &Table { + &self.table + } + + pub fn log(&self, line: &str) { + (self.log)(line) + } + + /// The owner's grants, printed once per distinct set of problems. + pub fn grants(&self) -> Grants { + match grants::load(&self.cfg.paths.grants) { + Ok(set) => { + let mut printed = self.printed.lock().unwrap_or_else(|p| p.into_inner()); + *printed = None; + Ok(set) + } + Err(problems) => { + let mut printed = self.printed.lock().unwrap_or_else(|p| p.into_inner()); + if printed.as_deref() != Some(problems.as_slice()) { + self.log(grants::render(&problems).trim_end()); + } + *printed = Some(problems.clone()); + Err(problems) + } + } + } +} + +/// The snake_case wire name of a message: all fourteen kinds, none with a leading underscore. +pub fn kind(msg: &Message) -> &'static str { + match msg { + Message::ToolRequest(_) => "tool_request", + Message::ToolResponse(_) => "tool_response", + Message::Error(_) => "error", + Message::Turn(_) => "turn", + Message::TurnEvent(_) => "turn_event", + Message::TurnDone(_) => "turn_done", + Message::Approvals(_) => "approvals", + Message::ApprovalList(_) => "approval_list", + Message::Approve(_) => "approve", + Message::ApproveResult(_) => "approve_result", + Message::Refuse(_) => "refuse", + Message::Ok(_) => "ok", + Message::CheckGrants(_) => "check_grants", + Message::GrantsReport(_) => "grants_report", + } +} + +/// One frame on the wire: the version, the request's id, whether it is the last, and the message. +/// True when the frame reached the peer. +pub fn send(stream: &mut UnixStream, id: u64, last: bool, msg: Message) -> bool { + let env = Envelope { + v: PROTOCOL_VERSION, + id, + r#final: last, + msg, + }; + write_frame(stream, &env).is_ok() +} + +/// The next frame, or `None` when the peer is gone. A malformed frame is answered with an error +/// frame and then `None`. +pub fn read_request(stream: &mut UnixStream) -> Option { + match read_frame(stream) { + Ok(env) => Some(env), + Err(FrameError::Closed) => None, + Err(frame_error) => { + let code = match &frame_error { + FrameError::BadVersion(_) => ErrorCode::BadVersion, + FrameError::Json(_) => ErrorCode::BadMessage, + _ => ErrorCode::BadFrame, + }; + let _ = send( + stream, + 0, + true, + Message::Error(WireError { + code, + detail: frame_error.to_string(), + }), + ); + None + } + } +} + +/// Refuse a message that does not belong on this socket: log it, then answer with `Forbidden`. +pub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message, socket: &str) { + let kind_name = kind(msg); + broker.log(&format!( + "brokerd: refused the message kind {kind_name} on {socket}\nsee docs/runbook.md#socket-forbidden" + )); + let _ = send( + stream, + id, + true, + Message::Error(WireError { + code: ErrorCode::Forbidden, + detail: format!("{kind_name} is not accepted on {socket}"), + }), + ); +} + +/// Whether the peer is still there: it sends nothing more and never half-closes, so a byte would +/// break the protocol and a timeout means it is waiting. +pub fn alive(stream: &UnixStream) -> bool { + let mut stream = stream; + if stream + .set_read_timeout(Some(Duration::from_millis(10))) + .is_err() + { + return false; + } + let mut byte = [0u8; 1]; + match stream.read(&mut byte) { + Ok(_) => false, + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => + { + true + } + Err(_) => false, + } +} + +/// Answer one call: decide, and either answer it or wait for an approval. +pub fn handle(mut stream: UnixStream, broker: &Broker) { + let Some(envelope) = read_request(&mut stream) else { + return; // 1. the peer is gone before it speaks + }; + let id = envelope.id; + let msg = envelope.msg; + let Message::ToolRequest(request) = msg else { + forbid(broker, &mut stream, id, &msg, "broker.sock"); // 2. not a tool request + return; + }; + + let now = Timestamp::now(); + let grants = broker.grants(); + let answer = match broker.ledger().decide(request, &grants, now) { + Decided::Denied(reason) => Some(ToolResponse::Denied { reason }), // 3a. denied + Decided::Allowed { decision, seq } => Some(run(broker, decision, seq)), // 3b. allowed + Decided::Ask { ask, seq, state } => { + pending(&mut stream, broker, id, ask, seq, state, now) // 3c. ask + } + }; + if let Some(answer) = answer { + let _ = send(&mut stream, id, true, Message::ToolResponse(answer)); // 4. the final frame + } +} + +/// Run an allowed call and record its result. +fn run(broker: &Broker, decision: crate::policy::Decision, seq: u64) -> ToolResponse { + let call = Call::of(&decision, seq); + let response = runner::run(decision, broker.runtime.as_ref()); + broker.ledger().finish(&call, response, Timestamp::now()) +} + +/// The `ask` path: send one pending frame, wait for the verdict, then answer. `None` means send +/// nothing more: the requester left while it was pending. +fn pending( + stream: &mut UnixStream, + broker: &Broker, + id: u64, + ask: Ask, + seq: u64, + state: SessionState, + now: Timestamp, +) -> Option { + // 1. The expiry: now plus the ttl, or the grant's own expiry if that is earlier. + let ttl = broker.cfg().approvals.ttl_ms; + let by_ttl = match Timestamp::from_unix_millis(now.unix_millis().saturating_add(ttl)) { + Ok(expires) => expires, + Err(_) => Timestamp::MAX, + }; + let expires = match ask.expires() { + Some(expiry) if expiry < by_ttl => expiry, + _ => by_ttl, + }; + + // 2. Record the pending call in the table. + let info = PendingApproval { + approval: seq, + session: ask.request().session.clone(), + call: ask.request().call, + tool: ask.request().tool.clone(), + arguments: ask.args().canonical_json(), + grant: ask.grant().to_string(), + taint: state.taint, + created: now, + expires, + }; + let verdict = broker.table().insert(info, ask); + + // 3. Show the caller the pending call. If the frame does not reach the peer, the entry is + // still ours to take; if it is gone, someone is already answering it. + if !send( + stream, + id, + false, + Message::ToolResponse(ToolResponse::PendingApproval { + approval: seq, + expires, + }), + ) && broker.table().take(seq).is_some() + { + return None; // 3. the peer left before the frame went out + } + + // 4. Wait for the verdict, or for the peer to leave. + let outcome = loop { + match verdict.recv_timeout(Duration::from_secs(1)) { + Ok(verdict) => break Ok(verdict), + Err(RecvTimeoutError::Timeout) => { + if alive(stream) { + continue; + } + match broker.table().take(seq) { + Some(_) => return None, // 4. the peer left while we waited + None => match verdict.recv() { + Ok(verdict) => break Ok(verdict), + Err(_) => break Err(DenyReason::AuditUnavailable), + }, + } + } + Err(RecvTimeoutError::Disconnected) => break Err(DenyReason::AuditUnavailable), // 4. the taker dropped it unanswered + } + }; + + // 5. Answer from the verdict. + match outcome { + Ok(verdict) => match verdict { + Verdict::Denied(reason) => Some(ToolResponse::Denied { reason }), + Verdict::Run(decision) => { + let decision = *decision; + if !alive(stream) { + let _ = broker.ledger().finish( + &Call::of(&decision, seq), + ToolResponse::Failed { + message: GONE.to_string(), + }, + now, + ); + return None; // 5. the peer left at the last look; nothing ran + } + Some(run(broker, decision, seq)) + } + }, + Err(reason) => Some(ToolResponse::Denied { reason }), + } +} diff --git a/crates/brokerd/src/lib.rs b/crates/brokerd/src/lib.rs index 692d85c..6edac64 100644 --- a/crates/brokerd/src/lib.rs +++ b/crates/brokerd/src/lib.rs @@ -3,6 +3,7 @@ pub mod approvals; pub mod args; pub mod audit; +pub mod broker; pub mod config; pub mod grants; pub mod ledger; diff --git a/crates/brokerd/tests/broker.rs b/crates/brokerd/tests/broker.rs new file mode 100644 index 0000000..5f5a20e --- /dev/null +++ b/crates/brokerd/tests/broker.rs @@ -0,0 +1,324 @@ +//! 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 client::{Serve, call, next, open}; +use proto::{ + ApprovalList, Approve, ApproveResult, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, + Envelope, ErrorCode, GrantsReport, Message, Refuse, ResultStatus, SessionId, ToolResponse, + Turn, TurnDone, TurnEvent, Usage, WireError, +}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +const NOTES: &str = r#"{"path":"/n/a"}"#; + +fn read_notes(call_id: u64) -> proto::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 } +} + +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/crates/brokerd/tests/broker_pending.rs b/crates/brokerd/tests/broker_pending.rs new file mode 100644 index 0000000..54e8093 --- /dev/null +++ b/crates/brokerd/tests/broker_pending.rs @@ -0,0 +1,230 @@ +//! 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 client::{Serve, next, open}; +use proto::{ + AuditEvent, DataClass, DecisionRecord, DenyReason, Message, ResultStatus, Timestamp, + ToolResponse, +}; +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()), + } +} + +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()), + reason: None, + }; + answer_by_hand(&broker, approval, refused); + assert_eq!( + final_answer(&mut stream), + ToolResponse::Denied { + reason: DenyReason::ApprovalRefused + } + ); + 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, proto::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/crates/brokerd/tests/broker_sequence.rs b/crates/brokerd/tests/broker_sequence.rs new file mode 100644 index 0000000..9976f1d --- /dev/null +++ b/crates/brokerd/tests/broker_sequence.rs @@ -0,0 +1,238 @@ +//! Properties over sequences of calls, through `broker::handle` with many threads at once: +//! the log verifies and no `seq` repeats; taint never goes down; every `Result` follows the +//! record that let its call run; the runtime sees a call only after `allowed` or an approval. +//! 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::collections::BTreeMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use brokerd::ledger::Answer; +use client::{Serve, call}; +use proto::{ + AuditEvent, AuditRecord, CallId, DataClass, DecisionRecord, Message, SessionId, Timestamp, + ToolResponse, +}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +fn grants(rig: &Rig) { + let with = |class: &str, max: &str| { + grant_text("read_file", "auto", "", "") + .replace( + "result_class = \"private\"", + &format!("result_class = \"{class}\""), + ) + .replace("max_taint = \"secret\"", &format!("max_taint = \"{max}\"")) + }; + rig.grant("notes", &(with("private", "secret") + "paths = [\"/n\"]\n")); + rig.grant("keys", &(with("secret", "secret") + "paths = [\"/k\"]\n")); + // Stops applying once a session has read a secret. + rig.grant( + "public", + &(with("public", "private") + "paths = [\"/p\"]\n"), + ); + rig.grant( + "never", + &grant_text("read_file", "deny", "", "paths = [\"/d\"]"), + ); + rig.grant( + "asked", + &grant_text("read_file", "ask", "", "paths = [\"/a\"]"), + ); +} + +/// Which call ran on what authority: (session, call) of every Decision that allowed and every +/// Approval whose re-decision let the call run, keyed by the decision's seq. +fn check_sequence(records: &[AuditRecord]) -> usize { + let mut may_run: BTreeMap = BTreeMap::new(); + let mut taint: BTreeMap = BTreeMap::new(); + let mut results = 0; + for (i, r) in records.iter().enumerate() { + assert_eq!(r.seq, i as u64, "no seq repeats or skips"); + match &r.event { + AuditEvent::Decision { + session, + call, + outcome: DecisionRecord::Allowed {}, + .. + } => { + may_run.insert(r.seq, (session.clone(), *call)); + } + AuditEvent::Approval { + session, + call, + decision, + outcome: DecisionRecord::Allowed {} | DecisionRecord::Ask {}, + .. + } => { + may_run.insert(*decision, (session.clone(), *call)); + } + AuditEvent::Result { + session, + call, + decision, + taint_after, + .. + } => { + results += 1; + let allowed = may_run.remove(decision); + assert_eq!( + allowed, + Some((session.clone(), *call)), + "seq {}: a Result with no record letting its call run", + r.seq + ); + let before = taint.insert(session.as_str().to_string(), *taint_after); + assert!( + before.is_none_or(|b| b <= *taint_after), + "seq {}: taint went down", + r.seq + ); + } + _ => {} + } + } + assert!( + may_run.is_empty(), + "calls allowed but never finished: {may_run:?}" + ); + results +} + +#[test] +fn eight_threads_two_sessions_fifty_calls_each() { + let rig = Rig::new("sequence-many"); + grants(&rig); + let rt = Recording::answering("content"); + let broker = rig.broker(&rt); + let paths = ["/n/x", "/k/x", "/p/x", "/d/x", "/none/x"]; + let threads: Vec<_> = (0..8u64) + .map(|t| { + let broker = Arc::clone(&broker); + std::thread::spawn(move || { + let session = if t % 2 == 0 { "s-even" } else { "s-odd" }; + for i in 0..50u64 { + let path = paths[((t + i) % 5) as usize]; + let args = format!(r#"{{"path":"{path}"}}"#); + let req = request(session, t * 1000 + i, "read_file", &args); + let frames = call(&broker, req); + assert_eq!(frames.len(), 1); + } + }) + }) + .collect(); + for t in threads { + t.join().unwrap(); + } + let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap(); + assert!(report.failure.is_none(), "{:?}", report.failure); + assert!(report.unfinished.is_empty()); + let records = rig.records(); + let decisions = records + .iter() + .filter(|r| matches!(r.event, AuditEvent::Decision { .. })) + .count(); + assert_eq!(decisions, 400); + let results = check_sequence(&records); + assert_eq!( + rt.count(), + results, + "the runtime saw exactly the allowed calls" + ); + assert!(results > 0 && results < 400); + // Both sessions read a secret. + for s in ["s-even", "s-odd"] { + let state = rig.state().read(&SessionId::new(s).unwrap()).unwrap(); + assert_eq!(state.taint, DataClass::Secret); + } +} + +#[test] +fn approved_and_refused_calls_run_only_after_an_approval() { + let rig = Rig::new("sequence-ask"); + grants(&rig); + let rt = Recording::answering("content"); + let broker = rig.broker(&rt); + // An owner who approves even ids and refuses odd ones, as fast as they appear. + let stop = Arc::new(AtomicBool::new(false)); + let owner = { + let (broker, stop) = (Arc::clone(&broker), Arc::clone(&stop)); + std::thread::spawn(move || { + while !stop.load(Ordering::SeqCst) { + for item in broker.table().list() { + let Some(entry) = broker.table().take(item.approval) else { + continue; + }; + let answer = if item.approval % 2 == 0 { + Answer::Approved { by: None } + } else { + Answer::Refused { + by: None, + reason: None, + } + }; + let grants = broker.grants(); + let done = broker.ledger().answer( + entry.ask, + item.approval, + answer, + &grants, + Timestamp::now(), + ); + entry.reply.send(done.verdict).unwrap(); + } + std::thread::sleep(Duration::from_millis(2)); + } + }) + }; + let threads: Vec<_> = (0..4u64) + .map(|t| { + let broker = Arc::clone(&broker); + std::thread::spawn(move || { + for i in 0..10u64 { + let path = if i % 2 == 0 { "/a/x" } else { "/n/x" }; + let args = format!(r#"{{"path":"{path}"}}"#); + let frames = call(&broker, request("s1", t * 100 + i, "read_file", &args)); + let last = frames.last().unwrap(); + assert!( + matches!( + &last.msg, + Message::ToolResponse( + ToolResponse::Result { .. } | ToolResponse::Denied { .. } + ) + ), + "{last:?}" + ); + } + }) + }) + .collect(); + for t in threads { + t.join().unwrap(); + } + stop.store(true, Ordering::SeqCst); + owner.join().unwrap(); + let records = rig.records(); + let approvals = records + .iter() + .filter(|r| matches!(r.event, AuditEvent::Approval { .. })) + .count(); + assert_eq!(approvals, 20, "one Approval for each of the twenty asks"); + let results = check_sequence(&records); + assert_eq!(rt.count(), results); + let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap(); + assert!(report.failure.is_none() && report.abandoned.is_empty()); +} diff --git a/crates/brokerd/tests/support/client.rs b/crates/brokerd/tests/support/client.rs new file mode 100644 index 0000000..65ef1e2 --- /dev/null +++ b/crates/brokerd/tests/support/client.rs @@ -0,0 +1,77 @@ +//! A client for `broker::handle` and `admin::handle` over a socket pair, and a `Broker` built on +//! a `Rig`. Do not edit. +//! +//! Included with `#[path = "support/client.rs"] mod client;`, beside `rig`, `runtime`, `sink` +//! and `tmp`. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::os::unix::net::UnixStream; +use std::sync::Arc; +use std::time::Duration; + +use brokerd::broker::{self, Broker}; +use proto::{Envelope, Message, PROTOCOL_VERSION, ToolRequest}; + +use crate::rig::Rig; +use crate::runtime::{Recording, Shared}; + +pub trait Serve { + /// A `Broker` on this rig's home, with `runtime` and the rig's flaky sink and log. + fn broker(&self, runtime: &Arc) -> Arc; +} + +impl Serve for Rig { + fn broker(&self, runtime: &Arc) -> Arc { + let runtime = Box::new(Shared(Arc::clone(runtime))); + Arc::new(Broker::new( + self.cfg.clone(), + self.ledger(), + runtime, + self.lines.sink(), + )) + } +} + +/// A connection to `handler` running on its own thread, with `msg` already sent under `id`. +pub fn open( + broker: &Arc, + handler: fn(UnixStream, &Broker), + id: u64, + msg: Message, +) -> UnixStream { + let (mut client, server) = UnixStream::pair().unwrap(); + let broker = Arc::clone(broker); + std::thread::spawn(move || handler(server, &broker)); + let env = Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg, + }; + proto::write_frame(&mut client, &env).unwrap(); + client +} + +/// The next frame, waiting at most ten seconds. +pub fn next(stream: &mut UnixStream) -> Envelope { + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + proto::read_frame(stream).unwrap() +} + +/// Sends a tool request to `broker::handle` and reads every frame up to the final one. +pub fn call(broker: &Arc, req: ToolRequest) -> Vec { + let id = req.call.0; + let mut stream = open(broker, broker::handle, id, Message::ToolRequest(req)); + let mut frames = Vec::new(); + loop { + let env = next(&mut stream); + let last = env.r#final; + frames.push(env); + if last { + return frames; + } + } +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 565e7b0..0abec73 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M3a/13-brokerd-broker | 2026-09-20 | done | 3 | fail | none | Wrote crates/brokerd/src/broker.rs (313 lines): Broker { cfg, ledger, table, runtime, log, printed: Mutex>> } with grants() printing each distinct set of problems once (recovering a poisoned lock with into_inner); kind (all fourteen snake_case wire names, no wildcard arm); send (write_frame(..).is_ok()); read_request (Closed->None, other errors send an error frame id 0 final true with BadVersion/BadMessage/BadFrame and the error text); forbid (log the runbook line, send Forbidden); alive (10ms read timeout, WouldBlock/TimedOut->true); handle (decide, then Denied/Allowed->run/Ask->pending, one final frame carrying the request id); run (Call::of, runner::run, ledger.finish); pending (expires = earlier of now+ttl or the grant's own expiry, insert the table entry, send one PendingApproval final:false, then wait on recv_timeout with the gone/taken lookups and AuditUnavailable). GONE is pub so the tests can name it. Every record goes through the ledger, nothing is written to it here. Three suites pass five runs in a row (9, 5, 2); the pending suite takes ~1s on its one-second look. First gate failed on fmt import order then clippy (three needless returns, one collapsible_if, one needless borrow on grants::render); fixed all and `make gate` prints `gate: ok`. | ? | | M3a/12-brokerd-ledger | 2026-09-20 | done | 2 | fail | none | Wrote crates/brokerd/src/ledger.rs (499 lines): Ledger + Inner { audit, state, stopped } behind one Mutex, and the three steps that hold it. decide copies the request out, reads state then policy::decide, and records the outcome (allowed/ask/denied, grant fields set together) as AuditEvent::Decision; answer re-decides an approval (approved only) and records AuditEvent::Approval with the answer/by/reason; finish raises the state for a Result and records AuditEvent::Result by its message otherwise, returning response unchanged only once the raised taint and the record are both on disk. Helpers not_recorded/audit_unavailable/denied; every append Err sets stopped through the one append method, and finish logs the raise error "brokerd: {e}" before stopping. Step 5 verified: each numbered exit points at a line and every append Err goes through the one stopped place. Trimmed 588 to 499 by compressing the module doc; one clippy fix (needless `return` in the answer append match, which is the tail expression). 11 + 9 tests pass; `make gate` prints `gate: ok`. | ? | | M3a/08-brokerd-state | 2026-09-18 | done | 1 | pass | none | Wrote crates/brokerd/src/state.rs: RUNBOOK, StateError (Unreadable/Write with hand-written Display ending in RUNBOOK and std::error::Error), StateStore (new does not touch disk, path joins /.json, read, raise) and the private StateFile with deny_unknown_fields. read has exactly one default path (ErrorKind::NotFound); Public taint is Unreadable; raise computes max(taint,label,Private) and ORs untrusted, always writes atomically in six steps mapping any error to Write(path, err). `cargo fmt` put `state` after `runner` in lib.rs. 9 tests pass; `make gate` prints `gate: ok`. | Laguna S 2.1 | | M1/01-workspace-and-gate | 2026-09-17 | done | 1 | pass | none | Crate skeletons, Cargo files and the given Makefile/deny.toml/test-gate-scripts.sh were already present untracked from a prior attempt; I verified them against the plan and created only the missing gate scripts, dependencies.md, egress.md and this log row. | Laguna S 2.1 |