boxmaker/crates/brokerd/tests/courier.rs

//! The courier carries a pending approval to `gatewayd` and its answer back, against a fake
//! `gatewayd` on a real socket (M4b spec, section 5). Do not edit.

#[path = "support/client.rs"]
mod client;
#[path = "support/fake_gateway.rs"]
mod fake_gateway;
#[path = "support/rig.rs"]
mod rig;
#[path = "support/runtime.rs"]
mod runtime;
#[path = "support/sink.rs"]
mod sink;
#[path = "support/tmp.rs"]
mod tmp;

use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::{Duration, Instant};

use brokerd::broker::{self, Broker};
use brokerd::config::Gateway;
use brokerd::courier::{self, Timing};
use brokerd::ledger::Answer;
use bxcore::{
    ApprovalAnswer, ApprovalReply, AuditEvent, DenyReason, GatewayOutcome, Message, Timestamp,
    ToolResponse,
};
use client::{Serve, next, open};
use fake_gateway::{FakeGateway, KYLE, POST};
use rig::{Rig, grant_text, request};
use runtime::Recording;

const WAIT: Duration = Duration::from_secs(5);

/// A rig whose `[gateway]` is a fake gatewayd in its home, and an `ask` grant for /n.
fn rig_with_gateway(tag: &str) -> (Rig, FakeGateway) {
    let mut rig = Rig::new(tag);
    let socket = rig.dir.path().join("gw.sock");
    let fake = FakeGateway::start(&socket);
    rig.cfg.gateway = Some(Gateway { socket });
    rig.grant(
        "notes",
        &grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
    );
    (rig, fake)
}

/// A call that waits for approval; its connection and the approval id.
fn ask(broker: &Arc<Broker>) -> (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<String>,
    Option<String>,
    Option<String>,
) {
    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<String> {
    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<T>(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 { .. }
    ));
}