Handle a tool request from decision to answer

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

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<dyn Fn(&str) + Send + Sync>;
+
+/// 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<dyn Runtime>,
+    log: Log,
+    printed: Mutex<Option<Vec<proto::GrantProblem>>>,
+}
+
+impl Broker {
+    pub fn new(cfg: Config, ledger: Ledger, runtime: Box<dyn Runtime>, 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<Envelope> {
+    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<ToolResponse> {
+    // 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 }),
+    }
+}

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;