boxmaker/crates/gatewayd/tests/gateway_sock.rs

//! `gateway.sock` from `gatewayd`'s side: the socket's permissions, and one connection's frames,
//! with the test playing both the serve loop and `brokerd` (M4b spec, sections 4 and 6). Do not
//! edit.

#[path = "support/tmp.rs"]
mod tmp;

use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::sync::mpsc::{self, Receiver};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use bxcore::{
    ApprovalOutcome, ApprovalReply, ApprovalRequest, CallId, DataClass, Envelope, ErrorCode,
    GatewayOutcome, Message, PROTOCOL_VERSION, PendingApproval, SessionId, Timestamp, read_frame,
    write_frame,
};
use gatewayd::gateway::{Ending, Timing, ToLoop, listen, serve};
use tmp::TempDir;

const KYLE: &str = "k0000000000000000000000000";
const POST: &str = "p0000000000000000000000001";
const CMD: &str = "c0000000000000000000000001";
const WAIT: Duration = Duration::from_secs(5);

fn info(approval: u64) -> PendingApproval {
    PendingApproval {
        approval,
        session: SessionId::new("mm-r0000000000000000000000001").unwrap(),
        call: CallId(3),
        tool: "shell".to_string(),
        arguments: "{}".to_string(),
        grant: "g".to_string(),
        taint: DataClass::Private,
        created: Timestamp::parse("2026-09-24T08:05:00.000Z").unwrap(),
        expires: Timestamp::parse("2026-09-24T08:20:00.000Z").unwrap(),
    }
}

fn frame(id: u64, last: bool, msg: Message) -> Envelope {
    Envelope {
        v: PROTOCOL_VERSION,
        id,
        r#final: last,
        msg,
    }
}

fn timing() -> Timing {
    Timing {
        first_frame: Duration::from_millis(500),
        posting: WAIT,
        outcome: Duration::from_millis(500),
        tick: Duration::from_millis(20),
    }
}

/// A listening gatewayd side: the socket and what it asks of the loop.
fn start(
    dir: &TempDir,
) -> (
    std::path::PathBuf,
    Receiver<ToLoop>,
    Arc<Mutex<Vec<String>>>,
) {
    let socket = dir.path().join("run/gateway/gateway.sock");
    let listener = listen(&socket).unwrap();
    let (tx, rx) = mpsc::channel();
    let lines = Arc::new(Mutex::new(Vec::new()));
    let sink = Arc::clone(&lines);
    serve(
        listener,
        tx,
        timing(),
        Arc::new(move |l: &str| sink.lock().unwrap().push(l.to_string())),
    )
    .unwrap();
    (socket, rx, lines)
}

/// brokerd: connect and send the request.
fn request(socket: &Path, approval: u64) -> UnixStream {
    let mut s = UnixStream::connect(socket).unwrap();
    s.set_read_timeout(Some(WAIT)).unwrap();
    write_frame(
        &mut s,
        &frame(
            1,
            true,
            Message::ApprovalRequest(ApprovalRequest {
                approval: info(approval),
            }),
        ),
    )
    .unwrap();
    s
}

fn reply(approval: u64) -> ApprovalReply {
    ApprovalReply {
        approval,
        approved: true,
        by: KYLE.to_string(),
        post: CMD.to_string(),
        reason: None,
    }
}

#[test]
fn the_socket_is_private_and_a_stale_one_is_replaced() {
    let dir = TempDir::new("gw-perms");
    let socket = dir.path().join("run/gateway/gateway.sock");
    drop(listen(&socket).unwrap());
    let listener = listen(&socket).expect("a stale socket is replaced");
    let mode = |p: &Path| std::fs::metadata(p).unwrap().permissions().mode() & 0o777;
    assert_eq!(
        (mode(&socket), mode(socket.parent().unwrap())),
        (0o600, 0o700)
    );
    drop(listener);
    let file = dir.path().join("run/gateway/not-a-socket");
    std::fs::write(&file, "x").unwrap();
    let why = listen(&file).unwrap_err();
    assert!(
        why.starts_with(&format!("cannot listen on {}", file.display())),
        "{why}"
    );
    assert!(
        dir.path().join("run/gateway/not-a-socket").exists(),
        "a file that is not a socket is left alone"
    );
}

#[test]
fn posted_answered_and_ended() {
    let dir = TempDir::new("gw-answer");
    let (socket, to_loop, _) = start(&dir);
    let mut brokerd = request(&socket, 42);
    let (answer, shown) = match to_loop.recv_timeout(WAIT).unwrap() {
        ToLoop::Show {
            info: got,
            answer,
            shown,
        } => {
            assert_eq!(got, info(42));
            (answer, shown)
        }
        other => panic!("{other:?}"),
    };
    shown.send(Ok(vec![POST.to_string()])).unwrap();
    let posted = read_frame(&mut brokerd).unwrap();
    assert_eq!((posted.id, posted.r#final), (1, false));
    assert!(
        matches!(posted.msg, Message::ApprovalPosted(ref p) if p.posts == [POST]),
        "{posted:?}"
    );
    answer.send(reply(42)).unwrap();
    let answered = read_frame(&mut brokerd).unwrap();
    assert_eq!(answered, frame(1, true, Message::ApprovalReply(reply(42))));
    write_frame(
        &mut brokerd,
        &frame(
            2,
            true,
            Message::ApprovalOutcome(ApprovalOutcome {
                outcome: GatewayOutcome::Runs {},
            }),
        ),
    )
    .unwrap();
    match to_loop.recv_timeout(WAIT).unwrap() {
        ToLoop::Ended { approval, ending } => {
            assert_eq!(approval, 42);
            assert_eq!(
                ending,
                Ending::Answered {
                    by: KYLE.to_string(),
                    reason: None,
                    outcome: GatewayOutcome::Runs {}
                }
            );
        }
        other => panic!("{other:?}"),
    }
}

#[test]
fn brokerd_closing_first_ends_it_as_no_longer_pending() {
    let dir = TempDir::new("gw-closed");
    let (socket, to_loop, _) = start(&dir);
    let mut brokerd = request(&socket, 7);
    let ToLoop::Show { shown, answer, .. } = to_loop.recv_timeout(WAIT).unwrap() else {
        panic!()
    };
    shown.send(Ok(vec![POST.to_string()])).unwrap();
    read_frame(&mut brokerd).unwrap();
    drop(brokerd);
    match to_loop.recv_timeout(WAIT).unwrap() {
        ToLoop::Ended { approval, ending } => assert_eq!((approval, ending), (7, Ending::NoLonger)),
        other => panic!("{other:?}"),
    }
    assert!(
        answer.send(reply(7)).is_err(),
        "nobody listens for an answer any more"
    );
}

#[test]
fn an_approval_that_cannot_be_posted_is_an_error_to_brokerd() {
    let dir = TempDir::new("gw-cannot");
    let (socket, to_loop, _) = start(&dir);
    let mut brokerd = request(&socket, 8);
    let ToLoop::Show { shown, .. } = to_loop.recv_timeout(WAIT).unwrap() else {
        panic!()
    };
    shown.send(Err("Mattermost is down".to_string())).unwrap();
    let got = read_frame(&mut brokerd).unwrap();
    assert_eq!((got.id, got.r#final), (1, true));
    match got.msg {
        Message::Error(e) => assert_eq!(
            (e.code, e.detail.as_str()),
            (ErrorCode::Internal, "Mattermost is down")
        ),
        other => panic!("{other:?}"),
    }
    assert!(
        to_loop.recv_timeout(Duration::from_millis(300)).is_err(),
        "nothing more for the loop"
    );
}

#[test]
fn no_outcome_after_the_answer_is_no_longer_pending_and_logged() {
    let dir = TempDir::new("gw-silent");
    let (socket, to_loop, lines) = start(&dir);
    let mut brokerd = request(&socket, 9);
    let ToLoop::Show { shown, answer, .. } = to_loop.recv_timeout(WAIT).unwrap() else {
        panic!()
    };
    shown.send(Ok(vec![POST.to_string()])).unwrap();
    read_frame(&mut brokerd).unwrap();
    answer.send(reply(9)).unwrap();
    read_frame(&mut brokerd).unwrap();
    match to_loop.recv_timeout(WAIT).unwrap() {
        ToLoop::Ended { approval, ending } => assert_eq!((approval, ending), (9, Ending::NoLonger)),
        other => panic!("{other:?}"),
    }
    assert_eq!(
        *lines.lock().unwrap(),
        ["gatewayd: approval 9: brokerd said nothing after the answer"]
    );
}

#[test]
fn anything_but_a_request_first_is_dropped() {
    let dir = TempDir::new("gw-bad");
    let (socket, to_loop, _) = start(&dir);
    for first in [
        frame(1, true, Message::ApprovalReply(reply(1))),
        frame(
            2,
            true,
            Message::ApprovalRequest(ApprovalRequest { approval: info(1) }),
        ),
        frame(
            1,
            false,
            Message::ApprovalRequest(ApprovalRequest { approval: info(1) }),
        ),
    ] {
        let mut s = UnixStream::connect(&socket).unwrap();
        s.set_read_timeout(Some(WAIT)).unwrap();
        write_frame(&mut s, &first).unwrap();
        assert!(
            read_frame(&mut s).is_err(),
            "closed without an answer: {first:?}"
        );
    }
    let silent = UnixStream::connect(&socket).unwrap();
    std::thread::sleep(Duration::from_millis(700));
    drop(silent);
    assert!(
        to_loop.recv_timeout(Duration::from_millis(300)).is_err(),
        "nothing reached the loop"
    );
}