diff --git a/crates/brokerd/src/broker.rs b/crates/brokerd/src/broker.rs index 06f9a23..d141a22 100644 --- a/crates/brokerd/src/broker.rs +++ b/crates/brokerd/src/broker.rs @@ -6,7 +6,7 @@ use std::io::Read; use std::os::unix::net::UnixStream; use std::sync::Mutex; -use std::sync::mpsc::RecvTimeoutError; +use std::sync::mpsc::{Receiver, RecvTimeoutError}; use std::time::Duration; use proto::{ @@ -16,6 +16,7 @@ use proto::{ use crate::approvals::{Table, Verdict}; use crate::config::Config; +use crate::courier; use crate::grants; use crate::ledger::{Call, Decided, Grants, Ledger}; use crate::policy::{Ask, SessionState}; @@ -166,6 +167,44 @@ pub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message, ); } +/// What waiting for the verdict came to. +enum Waited { + Verdict(Result), + PeerLeft, +} + +/// Wait for the verdict, or for the peer to leave. If the peer leaves and the entry is still in +/// the table, taking it ends the approval unanswered; if someone already took it, their verdict +/// is awaited. +fn wait_for_verdict( + broker: &Broker, + stream: &UnixStream, + verdict: &Receiver, + seq: u64, +) -> Waited { + loop { + match verdict.recv_timeout(Duration::from_secs(1)) { + Ok(verdict) => return Waited::Verdict(Ok(verdict)), + Err(RecvTimeoutError::Timeout) => { + if alive(stream) { + continue; + } + match broker.table().take(seq) { + Some(_) => return Waited::PeerLeft, + None => match verdict.recv() { + Ok(verdict) => return Waited::Verdict(Ok(verdict)), + Err(_) => return Waited::Verdict(Err(DenyReason::AuditUnavailable)), + }, + } + } + // The taker dropped it unanswered. + Err(RecvTimeoutError::Disconnected) => { + return Waited::Verdict(Err(DenyReason::AuditUnavailable)); + } + } + } +} + /// 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 { @@ -256,6 +295,7 @@ fn pending( created: now, expires, }; + let shown = info.clone(); 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 @@ -273,24 +313,14 @@ fn pending( 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 - } + // 4. Wait for the verdict, or for the peer to leave, with a courier carrying the approval to + // Mattermost meanwhile when `[gateway]` is set. + let waited = courier::alongside(broker, &shown, || { + wait_for_verdict(broker, stream, &verdict, seq) + }); + let outcome = match waited { + Waited::Verdict(outcome) => outcome, + Waited::PeerLeft => return None, // 4. the peer left while we waited }; // 5. Answer from the verdict. diff --git a/crates/brokerd/src/courier.rs b/crates/brokerd/src/courier.rs new file mode 100644 index 0000000..a47b4c7 --- /dev/null +++ b/crates/brokerd/src/courier.rs @@ -0,0 +1,276 @@ +//! The courier: one pending approval carried to `gatewayd` over `gateway.sock`, and its answer +//! back (M4b spec, section 5). It runs beside the connection thread that waits for the verdict, +//! and is one more taker of the table's entry, answering it exactly as `bxctl` does. + +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; +use std::time::{Duration, Instant}; + +use proto::{ + ApprovalOutcome, ApprovalReply, ApprovalRequest, DecisionRecord, DenyReason, Envelope, + FrameError, GatewayOutcome, Message, PendingApproval, Timestamp, read_frame, +}; + +use crate::admin; +use crate::broker::{Broker, send}; +use crate::ledger::Answer; + +pub const RUNBOOK: &str = "see docs/runbook.md#gateway-unreachable"; + +/// How long to wait before trying again, and how often to look at the table meanwhile. +#[derive(Debug, Clone, Copy)] +pub struct Timing { + pub retry: Duration, + pub tick: Duration, +} + +impl Default for Timing { + fn default() -> Self { + Timing { + retry: Duration::from_secs(5), + tick: Duration::from_secs(1), + } + } +} + +/// How one attempt ended. +enum Attempt { + /// The approval is answered or no longer pending: the courier is done. + Done, + /// It was never posted: `gatewayd` could not be reached, or could not post. + Unreachable(String), + /// It was posted, then `gatewayd` went away or broke the protocol. + Lost(String), +} + +/// Run `wait` (the connection thread's wait for the verdict) with a courier beside it, when +/// `[gateway]` is set. The courier stops when `wait` returns. +pub fn alongside(broker: &Broker, info: &PendingApproval, wait: impl FnOnce() -> T) -> T { + let Some(socket) = broker.cfg().gateway_socket() else { + return wait(); + }; + let stop = AtomicBool::new(false); + std::thread::scope(|scope| { + let started = std::thread::Builder::new().spawn_scoped(scope, || { + carry(broker, &socket, info, &stop, Timing::default()) + }); + if let Err(e) = started { + broker.log(&format!( + "brokerd: approval {}: cannot start its courier: {e}; bxctl can still answer it", + info.approval + )); + } + let out = wait(); + stop.store(true, Ordering::SeqCst); + out + }) +} + +/// Carry approval `info` to `gatewayd` until it is answered or no longer pending. +pub fn carry( + broker: &Broker, + socket: &Path, + info: &PendingApproval, + stop: &AtomicBool, + timing: Timing, +) { + let id = info.approval; + let mut warned = false; + loop { + if finished(broker, id, stop) { + return; + } + match attempt(broker, socket, info, stop, timing, warned) { + Attempt::Done => return, + Attempt::Unreachable(why) => { + if !warned { + warned = true; + broker.log(&format!( + "brokerd: approval {id}: gatewayd is not reachable at {}: {why}; bxctl can still answer it\n{RUNBOOK}", + socket.display() + )); + } + pause(broker, id, stop, timing); + } + Attempt::Lost(why) => { + broker.log(&format!("brokerd: approval {id}: {why}; posting it again")); + pause(broker, id, stop, timing); + } + } + } +} + +/// The courier's work is over: stopped, or the entry has left the table. +fn finished(broker: &Broker, id: u64, stop: &AtomicBool) -> bool { + stop.load(Ordering::SeqCst) || !broker.table().contains(id) +} + +/// Wait `timing.retry`, in ticks, unless the courier's work ends first. +fn pause(broker: &Broker, id: u64, stop: &AtomicBool, timing: Timing) { + let until = Instant::now() + timing.retry; + while Instant::now() < until && !finished(broker, id, stop) { + std::thread::sleep( + timing + .tick + .min(until.saturating_duration_since(Instant::now())), + ); + } +} + +/// Every frame `gatewayd` sends, read whole on a thread of its own, so the courier can wait with +/// a timeout without losing half a frame. Shutting the stream down ends the thread. +fn frames(stream: &UnixStream) -> std::io::Result>> { + // 1. `let mut reader = stream.try_clone()?;` and `let (tx, rx) = mpsc::channel();`. + // 2. `std::thread::Builder::new().spawn(move || loop { ... })?`, where the loop body is: `let + // frame = read_frame(&mut reader); let last = frame.is_err(); if tx.send(frame).is_err() || + // last { return; }`. + // 3. `Ok(rx)`. + let mut reader = stream.try_clone()?; + let (tx, rx) = mpsc::channel(); + std::thread::Builder::new().spawn(move || { + loop { + let frame = read_frame(&mut reader); + let last = frame.is_err(); + if tx.send(frame).is_err() || last { + return; + } + } + })?; + Ok(rx) +} + +/// One connection: post, then wait for the answer, looking at the table every tick. +fn attempt( + broker: &Broker, + socket: &Path, + info: &PendingApproval, + stop: &AtomicBool, + timing: Timing, + warned: bool, +) -> Attempt { + let mut stream = match UnixStream::connect(socket) { + Ok(stream) => stream, + Err(e) => return Attempt::Unreachable(e.to_string()), + }; + let rx = match frames(&stream) { + Ok(rx) => rx, + Err(e) => return Attempt::Unreachable(e.to_string()), + }; + let ended = exchange(broker, &mut stream, &rx, info, stop, timing, warned); + let _ = stream.shutdown(std::net::Shutdown::Both); + ended +} + +/// Frames 1 to 4 of section 4 on an open connection. +fn exchange( + broker: &Broker, + stream: &mut UnixStream, + rx: &Receiver>, + info: &PendingApproval, + stop: &AtomicBool, + timing: Timing, + warned: bool, +) -> Attempt { + let id = info.approval; + let request = Message::ApprovalRequest(ApprovalRequest { + approval: info.clone(), + }); + if !send(stream, 1, true, request) { + return Attempt::Unreachable("the request could not be sent".to_string()); + } + let mut posted = false; + loop { + let frame = match rx.recv_timeout(timing.tick) { + Ok(Ok(frame)) => frame, + Err(RecvTimeoutError::Timeout) => { + if finished(broker, id, stop) { + return Attempt::Done; + } + continue; + } + Ok(Err(_)) | Err(RecvTimeoutError::Disconnected) => { + return lost(posted, "gatewayd went away"); + } + }; + match (frame.id, frame.r#final, frame.msg) { + (1, false, Message::ApprovalPosted(_)) if !posted => { + posted = true; + if warned { + broker.log(&format!("brokerd: approval {id} posted in Mattermost")); + } + } + (1, true, Message::Error(e)) if !posted => { + return Attempt::Unreachable(format!("gatewayd could not post it: {}", e.detail)); + } + (1, true, Message::ApprovalReply(reply)) if posted => { + if reply.approval != id { + return lost(posted, "gatewayd answered another approval"); + } + if let Err(why) = reply.check() { + return lost(posted, &format!("gatewayd sent a bad answer ({why})")); + } + let outcome = answer(broker, id, reply); + let _ = send( + stream, + 2, + true, + Message::ApprovalOutcome(ApprovalOutcome { outcome }), + ); + return Attempt::Done; + } + _ => return lost(posted, "gatewayd broke the protocol"), + } + } +} + +fn lost(posted: bool, why: &str) -> Attempt { + if posted { + Attempt::Lost(why.to_string()) + } else { + Attempt::Unreachable(why.to_string()) + } +} + +/// Take the entry and answer it as `bxctl` would, with who answered and in which post. +fn answer(broker: &Broker, id: u64, reply: ApprovalReply) -> GatewayOutcome { + // 1. `let Some(entry) = broker.table().take(id) else { return GatewayOutcome::Gone {}; };` + // 2. `let (by, post) = (Some(reply.by), Some(reply.post));` and the `Answer`: `Answer::Approved + // { by, post }` when `reply.approved`, else `Answer::Refused { by, post, reason: + // reply.reason }`. + // 3. `match (reply.approved, admin::answer(broker, entry, answer, Timestamp::now()))`: (true, + // DecisionRecord::Denied { reason }) -> `GatewayOutcome::Denied { reason }`; (true, _) -> + // `GatewayOutcome::Runs {}`; (false, DecisionRecord::Denied { reason: + // DenyReason::ApprovalRefused }) -> `GatewayOutcome::Refused {}`; (false, + // DecisionRecord::Denied { reason }) -> `GatewayOutcome::Denied { reason }` (the refusal + // could not be recorded); (false, _) -> `GatewayOutcome::Refused {}`. + let Some(entry) = broker.table().take(id) else { + return GatewayOutcome::Gone {}; + }; + let (by, post) = (Some(reply.by), Some(reply.post)); + let answer = if reply.approved { + Answer::Approved { by, post } + } else { + Answer::Refused { + by, + post, + reason: reply.reason, + } + }; + match ( + reply.approved, + admin::answer(broker, entry, answer, Timestamp::now()), + ) { + (true, DecisionRecord::Denied { reason }) => GatewayOutcome::Denied { reason }, + (true, _) => GatewayOutcome::Runs {}, + ( + false, + DecisionRecord::Denied { + reason: DenyReason::ApprovalRefused, + }, + ) => GatewayOutcome::Refused {}, + (false, DecisionRecord::Denied { reason }) => GatewayOutcome::Denied { reason }, + (false, _) => GatewayOutcome::Refused {}, + } +} diff --git a/crates/brokerd/src/lib.rs b/crates/brokerd/src/lib.rs index f6769d9..ec384c0 100644 --- a/crates/brokerd/src/lib.rs +++ b/crates/brokerd/src/lib.rs @@ -7,6 +7,7 @@ pub mod audit; pub mod broker; pub mod config; pub mod container; +pub mod courier; pub mod grants; pub mod ledger; pub mod pipes; diff --git a/crates/brokerd/tests/courier.rs b/crates/brokerd/tests/courier.rs new file mode 100644 index 0000000..1b0207b --- /dev/null +++ b/crates/brokerd/tests/courier.rs @@ -0,0 +1,447 @@ +//! 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 client::{Serve, next, open}; +use fake_gateway::{FakeGateway, KYLE, POST}; +use proto::{ + ApprovalAnswer, ApprovalReply, AuditEvent, DenyReason, GatewayOutcome, Message, Timestamp, + ToolResponse, +}; +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 + } + ); + 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 + } + ); + 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/crates/brokerd/tests/support/fake_gateway.rs b/crates/brokerd/tests/support/fake_gateway.rs new file mode 100644 index 0000000..88e3149 --- /dev/null +++ b/crates/brokerd/tests/support/fake_gateway.rs @@ -0,0 +1,150 @@ +//! A fake `gatewayd` on a Unix socket: it accepts each courier's connection, reads the approval, +//! and lets the test post it, answer it, close, or read the outcome. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::mpsc::{self, Receiver}; +use std::time::Duration; + +use proto::{ + ApprovalPosted, ApprovalReply, Envelope, ErrorCode, GatewayOutcome, Message, PROTOCOL_VERSION, + PendingApproval, WireError, read_frame, write_frame, +}; + +pub const KYLE: &str = "k0000000000000000000000000"; +pub const POST: &str = "c0000000000000000000000001"; + +pub struct FakeGateway { + pub socket: PathBuf, + conns: Receiver, +} + +/// One courier's connection, its request already read. +pub struct Conn { + stream: UnixStream, + pub request: PendingApproval, +} + +fn frame(id: u64, last: bool, msg: Message) -> Envelope { + Envelope { + v: PROTOCOL_VERSION, + id, + r#final: last, + msg, + } +} + +impl FakeGateway { + /// Listen on `socket` (created now). + pub fn start(socket: &Path) -> FakeGateway { + let listener = UnixListener::bind(socket).unwrap(); + let (tx, conns) = mpsc::channel(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let Ok(env) = read_frame(&mut stream) else { + continue; + }; + let Message::ApprovalRequest(req) = env.msg else { + panic!("not a request: {env:?}") + }; + assert_eq!((env.id, env.r#final), (1, true)); + if tx + .send(Conn { + stream, + request: req.approval, + }) + .is_err() + { + return; + } + } + }); + FakeGateway { + socket: socket.to_path_buf(), + conns, + } + } + + /// The next courier connection within `wait`, or `None`. + pub fn next(&self, wait: Duration) -> Option { + self.conns.recv_timeout(wait).ok() + } +} + +impl Conn { + pub fn posted(&mut self) { + let posts = vec![POST.to_string()]; + write_frame( + &mut self.stream, + &frame(1, false, Message::ApprovalPosted(ApprovalPosted { posts })), + ) + .unwrap(); + } + + pub fn cannot_post(&mut self) { + let e = WireError { + code: ErrorCode::Internal, + detail: "Mattermost is down".to_string(), + }; + write_frame(&mut self.stream, &frame(1, true, Message::Error(e))).unwrap(); + } + + pub fn reply(&mut self, reply: ApprovalReply) { + let _ = write_frame( + &mut self.stream, + &frame(1, true, Message::ApprovalReply(reply)), + ); + } + + pub fn approve(&mut self) { + let approval = self.request.approval; + self.reply(ApprovalReply { + approval, + approved: true, + by: KYLE.to_string(), + post: POST.to_string(), + reason: None, + }); + } + + pub fn refuse(&mut self, reason: &str) { + let approval = self.request.approval; + self.reply(ApprovalReply { + approval, + approved: false, + by: KYLE.to_string(), + post: POST.to_string(), + reason: Some(reason.to_string()), + }); + } + + /// Frame 4 within five seconds, or `None` if the courier closed without one. + pub fn outcome(&mut self) -> Option { + self.stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + match read_frame(&mut self.stream) { + Ok(env) => { + assert_eq!((env.id, env.r#final), (2, true)); + match env.msg { + Message::ApprovalOutcome(o) => Some(o.outcome), + other => panic!("not an outcome: {other:?}"), + } + } + Err(_) => None, + } + } + + /// Did the courier close the connection within `wait`, without sending anything? + pub fn closed_within(&mut self, wait: Duration) -> bool { + self.stream.set_read_timeout(Some(wait)).unwrap(); + matches!(read_frame(&mut self.stream), Err(proto::FrameError::Closed)) + } + + pub fn close(self) { + let _ = self.stream.shutdown(std::net::Shutdown::Both); + } +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 108b0cf..a5734d3 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 | |---|---|---|---|---|---|---|---| +| M4b/06-brokerd-courier | 2026-09-24 | done | 1 | pass | none | Filled the four `todo!()`s in the copied `courier.rs`: `finished` (`stop.load(SeqCst) || !broker.table().contains(id)`); `pause` (a `while Instant::now() < until && !finished` loop sleeping `timing.tick.min(until.saturating_duration_since(now))`); `frames` (`try_clone` the stream, spawn a reader thread that hands whole frames over an mpsc channel so the courier can `recv_timeout` without losing half a frame, `Ok(rx)`); `answer` (`take` the entry → `Gone {}` if gone, build the `Answer` from `reply.approved`, call `admin::answer`, map the returned `DecisionRecord` through the five-arm match to `Runs`/`Denied`/`Refused`/`Denied`/`Refused`). The four `broker.rs` changes: `RecvTimeoutError` import widened to include `Receiver`, `use crate::courier;` after the `config` use, `let shown = info.clone();` before the insert, step 4 replaced by `courier::alongside(broker, &shown, || wait_for_verdict(...))` matching `Waited`, and the old step 4 moved to `wait_for_verdict` (with the `Waited` enum) before `pub fn alive`. `cargo build -p brokerd` clean; courier test went 1 pass/9 fail then 10 passed three times running; full brokerd suite green; `cargo fmt --all` then `make gate` prints `gate: ok` on the first run. | ? | | M4b/05-brokerd-config-gateway | 2026-09-24 | done | 1 | pass | none | Copied `docs/plans/M4b/files/crates/brokerd/tests/config_gateway.rs` and the new `support/rig.rs` (whose `Config` gained `gateway: None`) into `crates/brokerd/tests/`. In `config.rs`: added the `Gateway { socket: PathBuf }` struct (`#[serde(deny_unknown_fields, default)]`) before `default_podman`, the `gateway: Option` field (with `#[serde(default)]`) after `runner` in `Config`, and the `gateway_socket()` method before `audit_dir` — `None` when the section is absent, the default `/run/gateway/gateway.sock` when the socket is empty, else the path as given. `cargo fmt --all` then `cargo test -p brokerd` green with `config_gateway` 4 passed; `make gate` prints `gate: ok` on the first run. | ? | | M4b/04-brokerd-answer-post | 2026-09-24 | done | 1 | pass | none | Moved `enum Answer` and `struct Answered` from `ledger.rs` into `approvals.rs` before `pub struct Entry` (with doc comments), added `pub use crate::approvals::{Answer, Answered};` next to the `Verdict` use and `DecisionRecord` to the `use proto` line in `approvals.rs`. Added the `post` field to `Answer::Approved`/`Answer::Refused`, threaded it through `ledger::answer` (step 3 binds `post`, step 6 writes it into `AuditEvent::Approval` instead of `None`), and gave the two `bxctl` answers in `admin.rs` `post: None`. Added `Table::contains` before `take`. Copied `ledger_post.rs`, `broker_pending.rs`, `broker_sequence.rs` and `ledger_answer.rs` into `crates/brokerd/tests/`; every suite passes, `ledger_post` 3 passed, `ledger.rs` is 487 lines, and `make gate` prints `gate: ok` on the first run. | ? | | M4b/03-bxctl-deny | 2026-09-24 | done | 1 | pass | none | Copied `docs/plans/M4b/files/crates/bxctl/tests/cli_deny.rs` into `crates/bxctl/tests/`. In `parse`, `"refuse" => parse_refuse(rest, home)` became `"refuse" | "deny" => parse_refuse(rest, home)`, so `deny` produces the same `Command::Refuse` as `refuse` for every argument list including the error cases. Added a blank line and the two-line `bxctl deny [--reason TEXT] [--admin-socket PATH]` / `Another name for refuse.` block to `USAGE` after the `refuse` block, keeping the two-space and six-space indents. The copied test's `deny_parses_as_refuse` and `usage_names_deny` both failed before the changes and pass after; `cargo test -p bxctl` is green and `make gate` prints `gate: ok` on the first run. | ? | diff --git a/docs/runbook.md b/docs/runbook.md index e475d56..d778b0e 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -558,3 +558,37 @@ again would run it twice. Then send the message again in the thread. **Check.** A direct message to Boxmaker is answered. + +## gateway-unreachable + +**What you see.** `brokerd` prints `brokerd: approval : gatewayd is not reachable at : +; bxctl can still answer it`, then this entry, once per approval. No approval post +appears in Mattermost. `bxctl approvals` lists the approval as usual. + +**Why.** `brokerd.toml` has a `[gateway]` section, so each pending approval is carried to +`gatewayd` to be posted in Mattermost. The courier could not connect to `gateway.sock`, or +`gatewayd` answered that it could not post (Mattermost unreachable). The approval is not lost: it +stays pending, `bxctl` can answer it, and the courier tries again every 5 seconds until it is +posted or answered. When a later try works, `brokerd` prints `brokerd: approval posted in +Mattermost`. + +**Confirm.** + +```sh +ls -l # gatewayd creates it at start; [approvals] socket in gatewayd.toml +pgrep -x gatewayd # is it running? +``` + +and `gatewayd`'s own output: `cannot reach …` there means Mattermost is the problem (see +[mattermost-unreachable](#mattermost-unreachable)). + +**Fix.** Start `gatewayd`, or make `[gateway] socket` in `brokerd.toml` and `[approvals] socket` +in `gatewayd.toml` name the same path (both default to `/run/gateway/gateway.sock`). To turn +Mattermost approvals off, remove `[gateway]` from `brokerd.toml` and restart `brokerd`. + +**Check.** Trigger an `ask` call: its approval is posted in Mattermost, and `!approve ` in +its thread answers it. + +Where two Boxmakers share a channel thread, address approval commands to one of them +(`@ !approve `): a bare `!approve ` is taken by every Boxmaker with that id pending +in that thread.