Handle approvals, refusals and grant checks on admin.sock

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/brokerd/src/admin.rs

@@ -0,0 +1,129 @@
+//! One `admin.sock` connection: list, approve, refuse, check the grants; and expiry, which answers
+//! an approval the same way. Whoever takes an entry out of the table answers it.
+
+use std::os::unix::net::UnixStream;
+
+use crate::approvals::Entry;
+use crate::broker::{Broker, forbid, read_request, send};
+use crate::grants;
+use crate::ledger::{Answer, Answered};
+use proto::{
+    ApprovalList, Approve, ApproveResult, DecisionRecord, DenyReason, Empty, ErrorCode,
+    GrantsReport, Message, Refuse, Timestamp, WireError,
+};
+
+/// The Approval record's `by` for answers through admin.sock.
+pub const BY: &str = "bxctl";
+/// The refusal `bxctl` sees when the ledger could not record it.
+pub const REFUSAL_INTERNAL: &str =
+    "the refusal could not be recorded; the call is denied; see docs/runbook.md#audit-unavailable";
+
+/// Whoever takes an entry out of the table answers it: the ledger records the answer, then the
+/// verdict goes to the waiting thread, and only then is the record returned.
+pub fn answer(broker: &Broker, entry: Entry, answer: Answer, now: Timestamp) -> DecisionRecord {
+    // 1. The grants as they are now: an approval decides again with them.
+    let grants = broker.grants();
+
+    // 2. Record the answer; the ledger writes the Approval record and returns the verdict and
+    //    outcome together, or the audit-unavailable denial when it could not record.
+    let Entry { info, ask, reply } = entry;
+    let Answered { verdict, outcome } =
+        broker
+            .ledger()
+            .answer(ask, info.approval, answer, &grants, now);
+
+    // 3. The record is written either way; if the waiting thread is gone, there is nothing else to do.
+    if reply.send(verdict).is_err() {
+        broker.log(&format!(
+            "brokerd: approval {} was answered after its requester had gone",
+            info.approval
+        ));
+    }
+
+    // 4. The outcome the caller reports to bxctl.
+    outcome
+}
+
+/// Answer every approval whose time has run out, the same way an owner's refusal would.
+pub fn expire_due(broker: &Broker, now: Timestamp) -> usize {
+    let entries = broker.table().take_expired(now);
+    let count = entries.len();
+    for entry in entries {
+        answer(broker, entry, Answer::Expired, now);
+    }
+    count
+}
+
+/// Serve one `admin.sock` connection from bxctl: one request, one final answer.
+pub fn handle(mut stream: UnixStream, broker: &Broker) {
+    // 1. The peer is gone before it speaks, or its frame is malformed: answer and return.
+    let Some(envelope) = read_request(&mut stream) else {
+        return;
+    };
+    let id = envelope.id;
+    let msg = envelope.msg;
+    let now = Timestamp::now();
+
+    // 2. By message kind. `take` is the only way in: no one looks at an entry and removes it later.
+    let response = match msg {
+        Message::Approvals(_) => Message::ApprovalList(ApprovalList {
+            items: broker.table().list(),
+        }),
+        Message::Approve(Approve { approval }) => match broker.table().take(approval) {
+            None => Message::Error(WireError {
+                code: ErrorCode::NoSuchApproval,
+                detail: format!("approval {approval} is not pending"),
+            }),
+            Some(entry) => {
+                let outcome = answer(
+                    broker,
+                    entry,
+                    Answer::Approved {
+                        by: Some(BY.to_string()),
+                    },
+                    now,
+                );
+                Message::ApproveResult(ApproveResult { outcome })
+            }
+        },
+        Message::Refuse(Refuse { approval, reason }) => match broker.table().take(approval) {
+            None => Message::Error(WireError {
+                code: ErrorCode::NoSuchApproval,
+                detail: format!("approval {approval} is not pending"),
+            }),
+            Some(entry) => match answer(
+                broker,
+                entry,
+                Answer::Refused {
+                    by: Some(BY.to_string()),
+                    reason,
+                },
+                now,
+            ) {
+                DecisionRecord::Denied {
+                    reason: DenyReason::ApprovalRefused,
+                } => Message::Ok(Empty {}),
+                // The refusal could not be recorded: bxctl gets an Internal, the waiting call a denial.
+                _ => Message::Error(WireError {
+                    code: ErrorCode::Internal,
+                    detail: REFUSAL_INTERNAL.to_string(),
+                }),
+            },
+        },
+        // bxctl grants check shows the problems itself and must not use up the print-once of the broker.
+        Message::CheckGrants(_) => {
+            let problems = match grants::load(&broker.cfg().paths.grants) {
+                Ok(_) => Vec::new(),
+                Err(problems) => problems,
+            };
+            Message::GrantsReport(GrantsReport { problems })
+        }
+        other => {
+            forbid(broker, &mut stream, id, &other, "admin.sock");
+            return;
+        }
+    };
+
+    // 3. The final frame; a failed send is ignored.
+    let _ = send(&mut stream, id, true, response);
+}

crates/brokerd/src/lib.rs

@@ -1,5 +1,6 @@
 //! The broker: the only role that holds authority.
 
+pub mod admin;
 pub mod approvals;
 pub mod args;
 pub mod audit;

crates/brokerd/tests/admin.rs

@@ -0,0 +1,385 @@
+//! 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 client::{Serve, next, open};
+use proto::{
+    ApprovalAnswer, Approve, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, ErrorCode,
+    Message, Refuse, SessionId, Timestamp, ToolResponse,
+};
+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<Broker>, 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<Broker>, 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<Broker>, approval: u64) -> Message {
+    admin(broker, Message::Approve(Approve { approval }))
+}
+
+fn refuse(broker: &Arc<Broker>, 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<ErrorCode> {
+    match msg {
+        Message::Error(e) => Some(e.code),
+        _ => None,
+    }
+}
+
+fn denied(reason: DenyReason) -> ToolResponse {
+    ToolResponse::Denied { reason }
+}
+
+#[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<Message> = 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"
+    );
+}