dotfiles/rust/signal/tests/send.rs

#![cfg(test)]
#![cfg(target_os = "linux")]

use std::{
    cell::RefCell,
    io::{self, BufRead, BufReader, Write},
    os::{fd::AsRawFd, unix::net::UnixStream},
    thread,
    time::{Duration, Instant},
};

use serde_json::{Value, json};
use signal::Signal;

thread_local! {
    static CONNECTION: RefCell<Option<UnixStream>> = const { RefCell::new(None) };
}

// Patch libc's connect only in this integration-test executable. Each test
// supplies its own socket pair; no call can reach the real Signal daemon.
// SAFETY: This symbol uses libc's connect ABI and preserves its return/errno
// convention. It never delegates to a real connection attempt.
#[unsafe(no_mangle)]
unsafe extern "C" fn connect(
    socket: libc::c_int,
    address: *const libc::sockaddr,
    length: libc::socklen_t,
) -> libc::c_int {
    let path_offset = std::mem::offset_of!(libc::sockaddr_un, sun_path);
    let expected = b"/run/signal-cli/socket\0";
    if !address.is_null()
        && let Some(path_end) = path_offset.checked_add(expected.len())
        && usize::try_from(length).is_ok_and(|length| length >= path_end)
    {
        // SAFETY: connect's caller supplies an address valid for length bytes.
        // The bounds check above covers the entire pathname slice.
        let path = unsafe {
            std::slice::from_raw_parts(address.cast::<u8>().add(path_offset), expected.len())
        };
        if path == expected
            && let Some(stream) = CONNECTION.with_borrow_mut(Option::take)
        {
            // SAFETY: Both descriptors are live and owned by this thread.
            // dup2 replaces the unconnected socket with the test endpoint.
            return if unsafe { libc::dup2(stream.as_raw_fd(), socket) } < 0 {
                -1
            } else {
                0
            };
        }
    }
    // SAFETY: __errno_location returns this thread's valid errno pointer.
    unsafe { *libc::__errno_location() = libc::ECONNREFUSED };
    -1
}

fn test_connection() -> UnixStream {
    let (client, server) = UnixStream::pair().unwrap();
    CONNECTION.with_borrow_mut(|connection| assert!(connection.replace(client).is_none()));
    server
}

fn signal() -> Signal {
    Signal {
        from_phone: "1234567890".into(),
    }
}

#[test]
fn reuses_sender_for_different_recipients_and_ignores_unrelated_responses() {
    let client = signal();
    for (to_phone, body) in [
        ("0987654321", "failure\nwith details"),
        ("+15555550102", "another failure"),
    ] {
        let server = test_connection();
        let task = thread::spawn(move || {
            let mut reader = BufReader::new(server);
            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            let request: Value = serde_json::from_str(&line).unwrap();
            assert_eq!(
                request,
                json!({
                    "jsonrpc": "2.0", "method": "send", "id": "signal-alert",
                    "params": { "account": "1234567890", "recipient": [to_phone], "message": body },
                })
            );
            reader
                .get_mut()
                .write_all(b"{\"method\":\"receive\"}\n{\"id\":\"other\"}\n{\"id\":\"signal-")
                .unwrap();
            reader
                .get_mut()
                .write_all(b"alert\",\"result\":{}}\n")
                .unwrap();
        });
        client.send(to_phone, body).unwrap();
        task.join().unwrap();
    }
}

#[test]
fn rejects_rpc_errors_invalid_json_and_premature_close() {
    for (reply, expected) in [
        (
            "{\"id\":\"signal-alert\",\"error\":{\"code\":-1}}\n",
            "Signal RPC returned an error",
        ),
        ("invalid\n", "Invalid Signal RPC response"),
        ("", "socket closed"),
        ("{\"id\":\"signal-alert\"}", "socket closed"),
    ] {
        let server = test_connection();
        let task = thread::spawn(move || {
            let mut reader = BufReader::new(server);
            reader.read_line(&mut String::new()).unwrap();
            reader.get_mut().write_all(reply.as_bytes()).unwrap();
        });
        let error = signal().send("0987654321", "failure").unwrap_err();
        assert!(format!("{error:#}").contains(expected), "{error:#}");
        task.join().unwrap();
    }
}

#[test]
fn reports_connection_failure() {
    let error = signal().send("0987654321", "failure").unwrap_err();
    assert_eq!(error.to_string(), "Failed to connect to Signal RPC socket");
    assert_eq!(
        error.downcast_ref::<io::Error>().unwrap().kind(),
        io::ErrorKind::ConnectionRefused
    );
}

#[test]
fn unresponsive_server_times_out() {
    let _server = test_connection();
    let start = Instant::now();
    let error = signal().send("0987654321", "failure").unwrap_err();
    assert_eq!(error.to_string(), "Failed to read Signal RPC response");
    assert!(matches!(
        error.downcast_ref::<io::Error>().unwrap().kind(),
        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
    ));
    assert!(start.elapsed() >= Duration::from_secs(4));
}