use super::*;
use crate::test_support::remove;
fn short_directory(name: &str) -> std::path::PathBuf {
let scratch = std::path::PathBuf::from(std::env::var_os("HOME").expect("a home directory")).join("temp/agent");
let directory = scratch.join(format!("dbus-{}-{name}", std::process::id()));
std::fs::create_dir_all(&directory).expect("scratch directory");
return directory;
}
use std::os::unix::net::{UnixListener, UnixStream};
fn hand_field(code: u8, kind: u8, text: &str) -> Vec<u8> {
let mut bytes: Vec<u8> = vec![code, 1, kind, 0];
bytes.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
bytes.extend(text.as_bytes());
bytes.push(0);
return bytes;
}
fn pad8(bytes: &mut Vec<u8>) {
while bytes.len() % 8 != 0 {
bytes.push(0);
}
}
fn hand_message(kind: u8, serial: u32, fields: &[Vec<u8>], body: &[u8]) -> Vec<u8> {
let mut array: Vec<u8> = Vec::new();
for field in fields {
pad8(&mut array);
array.extend(field);
}
let mut message: Vec<u8> = vec![b'l', kind, 0, 1];
message.extend(u32::try_from(body.len()).expect("short").to_le_bytes());
message.extend(serial.to_le_bytes());
message.extend(u32::try_from(array.len()).expect("short").to_le_bytes());
message.extend(array);
pad8(&mut message);
message.extend(body);
return message;
}
#[test]
fn a_call_without_arguments_has_the_specified_layout() {
let wanted: Vec<u8> = hand_message(
METHOD_CALL,
1,
&[
hand_field(FIELD_PATH, b'o', MANAGER_PATH),
hand_field(FIELD_DESTINATION, b's', "org.freedesktop.systemd1"),
hand_field(FIELD_INTERFACE, b's', MANAGER_INTERFACE),
hand_field(FIELD_MEMBER, b's', "Reload"),
],
&[],
);
assert_eq!(encode_call(1, &RELOAD, &[]), wanted);
assert_eq!(wanted.len(), 160);
assert_eq!(u32::from_le_bytes([wanted[12], wanted[13], wanted[14], wanted[15]]), 143);
}
#[test]
fn a_call_with_two_strings_carries_them_in_the_body() {
let message: Vec<u8> = encode_call(7, &START_UNIT, &["cli-git.socket", "replace"]);
let mut body: Vec<u8> = Vec::new();
for text in ["cli-git.socket", "replace"] {
while body.len() % 4 != 0 {
body.push(0);
}
body.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
body.extend(text.as_bytes());
body.push(0);
}
let split: usize = message.len() - body.len();
assert_eq!(split % 8, 0, "the body starts on an eight-byte boundary");
assert_eq!(&message[split..], body.as_slice());
assert_eq!(u32::from_le_bytes([message[4], message[5], message[6], message[7]]) as usize, body.len());
assert_eq!(u32::from_le_bytes([message[8], message[9], message[10], message[11]]), 7);
let signature: Vec<u8> = vec![FIELD_SIGNATURE, 1, b'g', 0, 2, b's', b's', 0];
assert!(message.windows(signature.len()).any(|window| { return window == signature.as_slice(); }));
}
fn serial_field(serial: u32) -> Vec<u8> {
let mut field: Vec<u8> = vec![FIELD_REPLY_SERIAL, 1, b'u', 0];
field.extend(serial.to_le_bytes());
return field;
}
#[test]
fn replies_built_by_hand_are_read() {
let returned: Vec<u8> = hand_message(METHOD_RETURN, 2, &[serial_field(5)], &[]);
assert_eq!(decode_reply(&returned), Ok(Reply { kind: METHOD_RETURN, reply_serial: Some(5), error_name: None, first_text: None }));
let mut body: Vec<u8> = Vec::new();
body.extend(7u32.to_le_bytes());
body.extend(b"no room\0");
let failed: Vec<u8> = hand_message(
ERROR,
3,
&[hand_field(FIELD_ERROR_NAME, b's', "org.freedesktop.systemd1.NoSuchUnit"), serial_field(9), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]],
&body,
);
assert_eq!(
decode_reply(&failed),
Ok(Reply {
kind: ERROR,
reply_serial: Some(9),
error_name: Some(String::from("org.freedesktop.systemd1.NoSuchUnit")),
first_text: Some(String::from("no room"))
})
);
}
#[test]
fn damaged_replies_are_refused() {
let whole: Vec<u8> = hand_message(METHOD_RETURN, 2, &[serial_field(1)], &[]);
assert!(matches!(decode_reply(&whole[..10]), Err(CallError::Malformed(_))));
let mut big_endian: Vec<u8> = whole.clone();
big_endian[0] = b'B';
assert!(matches!(decode_reply(&big_endian), Err(CallError::Malformed(_))));
let mut lies: Vec<u8> = whole.clone();
lies[4..8].copy_from_slice(&1000u32.to_le_bytes());
assert!(matches!(decode_reply(&lies), Err(CallError::Malformed(_))));
let mut long_fields: Vec<u8> = whole;
long_fields[12..16].copy_from_slice(&u32::MAX.to_le_bytes());
assert!(matches!(decode_reply(&long_fields), Err(CallError::Malformed(_))));
}
#[test]
fn the_reader_is_total_over_damaged_input() {
let mut body: Vec<u8> = Vec::new();
body.extend(3u32.to_le_bytes());
body.extend(b"abc\0");
let samples: Vec<Vec<u8>> = vec![
hand_message(METHOD_RETURN, 2, &[serial_field(1)], &[]),
hand_message(ERROR, 3, &[hand_field(FIELD_ERROR_NAME, b's', "a.b.C"), serial_field(2), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]], &body),
];
for sample in samples {
for length in 0..=sample.len() {
let _ = decode_reply(&sample[..length]);
}
for at in 0..sample.len() {
for value in [0u8, 1, 0x7f, 0x80, 0xff] {
let mut changed: Vec<u8> = sample.clone();
changed[at] = value;
let _ = decode_reply(&changed);
}
}
}
}
#[test]
fn the_session_bus_path_is_found() {
let runtime: &Path = Path::new("/run/user/1000");
assert_eq!(session_bus_path(Some("unix:path=/run/user/1000/bus"), None), Some(std::path::PathBuf::from("/run/user/1000/bus")));
assert_eq!(
session_bus_path(Some("unix:abstract=/tmp/x,guid=1;unix:guid=2,path=/tmp/other"), None),
Some(std::path::PathBuf::from("/tmp/other"))
);
assert_eq!(session_bus_path(Some("unix:abstract=/tmp/x"), Some(runtime)), Some(runtime.join("bus")));
assert_eq!(session_bus_path(Some("unix:path=/a%20b"), Some(runtime)), Some(runtime.join("bus")));
assert_eq!(session_bus_path(None, Some(runtime)), Some(runtime.join("bus")));
assert_eq!(session_bus_path(None, None), None);
}
struct Script {
answers: Vec<Vec<Vec<u8>>>,
}
fn serve_bus(listener: UnixListener, script: Script, seen: std::sync::mpsc::Sender<(String, Vec<Vec<u8>>)>) {
std::thread::spawn(move || {
let (mut stream, first) = accept_and_authenticate(&listener);
let mut calls: Vec<Vec<u8>> = Vec::new();
for answers in script.answers {
calls.push(read_call(&mut stream));
for answer in answers {
stream.write_all(&answer).expect("answer");
}
}
let _sent = seen.send((first, calls));
});
}
fn accept_and_authenticate(listener: &UnixListener) -> (UnixStream, String) {
let (mut stream, _) = listener.accept().expect("a client");
let mut one: [u8; 1] = [0];
let mut first: Vec<u8> = Vec::new();
while !first.ends_with(b"\r\n") {
stream.read_exact(&mut one).expect("auth bytes");
first.push(one[0]);
}
stream.write_all(b"OK 00112233445566778899aabbccddeeff\r\n").expect("ok");
let mut begin: Vec<u8> = Vec::new();
while !begin.ends_with(b"\r\n") {
stream.read_exact(&mut one).expect("begin bytes");
begin.push(one[0]);
}
return (stream, String::from_utf8_lossy(&first).into_owned());
}
fn read_call(stream: &mut UnixStream) -> Vec<u8> {
let mut header: Vec<u8> = vec![0; 16];
stream.read_exact(&mut header).expect("header");
let body: usize = u32::from_le_bytes([header[4], header[5], header[6], header[7]]) as usize;
let fields: usize = u32::from_le_bytes([header[12], header[13], header[14], header[15]]) as usize;
let mut rest: Vec<u8> = vec![0; (16 + fields).div_ceil(8) * 8 + body - 16];
stream.read_exact(&mut rest).expect("rest");
header.extend(rest);
return header;
}
fn name_acquired() -> Vec<u8> {
let mut body: Vec<u8> = Vec::new();
body.extend(5u32.to_le_bytes());
body.extend(b":1.42\0");
return hand_message(4, 1, &[hand_field(FIELD_MEMBER, b's', "NameAcquired"), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]], &body);
}
fn string_return(serial: u32, text: &str) -> Vec<u8> {
let mut body: Vec<u8> = Vec::new();
body.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
body.extend(text.as_bytes());
body.push(0);
return hand_message(METHOD_RETURN, serial + 10, &[serial_field(serial), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]], &body);
}
fn path_return(serial: u32, text: &str) -> Vec<u8> {
let mut body: Vec<u8> = Vec::new();
body.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
body.extend(text.as_bytes());
body.push(0);
return hand_message(METHOD_RETURN, serial + 10, &[serial_field(serial), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b'o', 0]], &body);
}
fn variant_return(serial: u32, inner: u8, text: &str) -> Vec<u8> {
let mut body: Vec<u8> = vec![1, inner, 0, 0];
body.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
body.extend(text.as_bytes());
body.push(0);
return hand_message(METHOD_RETURN, serial + 10, &[serial_field(serial), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b'v', 0]], &body);
}
fn error_return(serial: u32, name: &str, text: &str) -> Vec<u8> {
let mut body: Vec<u8> = Vec::new();
body.extend(u32::try_from(text.len()).expect("short").to_le_bytes());
body.extend(text.as_bytes());
body.push(0);
return hand_message(ERROR, serial + 10, &[hand_field(FIELD_ERROR_NAME, b's', name), serial_field(serial), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]], &body);
}
#[test]
fn an_object_path_and_a_string_variant_are_read() {
assert_eq!(decode_reply(&path_return(4, "/org/freedesktop/systemd1/unit/a_2eservice")).expect("read").first_text.as_deref(), Some("/org/freedesktop/systemd1/unit/a_2eservice"));
let variant: Vec<u8> = variant_return(4, b's', "deactivating");
assert_eq!(decode_reply(&variant).expect("read").first_text.as_deref(), Some("deactivating"));
assert_eq!(decode_reply(&variant_return(4, b'u', "xxxx")).expect("read").first_text, None);
for length in 0..variant.len() {
let _ = decode_reply(&variant[..length]);
}
for at in 0..variant.len() {
for value in [0u8, 1, 0x7f, 0x80, 0xff] {
let mut changed: Vec<u8> = variant.clone();
changed[at] = value;
let _ = decode_reply(&changed);
}
}
}
const STATE_TEXT: &str = "deactivating";
fn first_text(message: &[u8]) -> Option<String> {
return decode_reply(message).expect("a message").first_text;
}
fn string_signature(end: u8) -> Vec<u8> {
return vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', end];
}
#[test]
fn a_body_string_without_its_end_byte_is_not_read() {
let intact: Vec<u8> = string_return(4, STATE_TEXT);
assert_eq!(first_text(&intact).as_deref(), Some(STATE_TEXT));
let mut unterminated: Vec<u8> = intact;
let end: usize = unterminated.len() - 1;
unterminated[end] = b'X';
assert_eq!(first_text(&unterminated), None);
}
#[test]
fn a_body_string_whose_end_byte_lies_outside_the_declared_body_is_not_read() {
let mut body: Vec<u8> = Vec::new();
body.extend(u32::try_from(STATE_TEXT.len()).expect("short").to_le_bytes());
body.extend(STATE_TEXT.as_bytes());
let mut outside: Vec<u8> = hand_message(METHOD_RETURN, 14, &[serial_field(4), string_signature(0)], &body);
outside.push(0);
assert_eq!(first_text(&outside), None);
}
#[test]
fn a_variant_without_an_end_byte_is_not_read() {
let variant: Vec<u8> = variant_return(4, b's', STATE_TEXT);
assert_eq!(first_text(&variant).as_deref(), Some(STATE_TEXT));
let mut value_unterminated: Vec<u8> = variant.clone();
let value_end: usize = value_unterminated.len() - 1;
value_unterminated[value_end] = b'X';
assert_eq!(first_text(&value_unterminated), None, "a variant string whose end byte is another byte");
let body_length: usize = 4 + 4 + STATE_TEXT.len() + 1;
let body_start: usize = variant.len() - body_length;
let mut signature_unterminated: Vec<u8> = variant;
signature_unterminated[body_start + 2] = b'X';
assert_eq!(first_text(&signature_unterminated), None, "a variant signature whose end byte is another byte");
}
#[test]
fn a_signature_field_without_its_end_byte_is_not_obeyed() {
let mut body: Vec<u8> = Vec::new();
body.extend(u32::try_from(STATE_TEXT.len()).expect("short").to_le_bytes());
body.extend(STATE_TEXT.as_bytes());
body.push(0);
let ended: Vec<u8> = hand_message(METHOD_RETURN, 14, &[serial_field(4), string_signature(0)], &body);
assert_eq!(first_text(&ended).as_deref(), Some(STATE_TEXT));
let unended: Vec<u8> = hand_message(METHOD_RETURN, 14, &[serial_field(4), string_signature(b'X')], &body);
assert_eq!(first_text(&unended), None);
}
#[test]
fn the_state_of_a_unit_is_read_from_its_object_and_an_unloaded_unit_has_none() {
let root = short_directory("state");
let socket = root.join("bus");
let listener = UnixListener::bind(&socket).expect("listen");
let (sender, receiver) = std::sync::mpsc::channel();
let script: Script = Script {
answers: vec![
vec![string_return(1, ":1.42")],
vec![path_return(2, "/org/freedesktop/systemd1/unit/cli_2dgit_2eservice")],
vec![variant_return(3, b's', "deactivating")],
vec![error_return(4, NO_SUCH_UNIT, "not loaded")],
],
};
serve_bus(listener, script, sender);
let mut bus: Bus = Bus::connect(socket.as_path(), Duration::from_secs(10)).expect("connected");
assert_eq!(bus.active_state("cli-git.service"), Ok(Some(String::from("deactivating"))));
assert_eq!(bus.active_state("cli-git.socket"), Ok(None));
let (_auth, calls) = receiver.recv().expect("seen");
let property: Target<'_> = Target {
destination: MANAGER_NAME,
path: "/org/freedesktop/systemd1/unit/cli_2dgit_2eservice",
interface: "org.freedesktop.DBus.Properties",
member: "Get",
};
assert_eq!(calls[1], encode_call(2, &GET_UNIT, &["cli-git.service"]));
assert_eq!(calls[2], encode_call(3, &property, &["org.freedesktop.systemd1.Unit", "ActiveState"]));
assert_eq!(calls[3], encode_call(4, &GET_UNIT, &["cli-git.socket"]));
remove(root.as_path());
}
#[test]
fn a_session_says_hello_then_reloads_and_starts() {
let root = short_directory("exchange");
let socket = root.join("bus");
let listener = UnixListener::bind(&socket).expect("listen");
let (sender, receiver) = std::sync::mpsc::channel();
let script: Script = Script {
answers: vec![
vec![name_acquired(), string_return(1, ":1.42")],
vec![hand_message(METHOD_RETURN, 12, &[serial_field(2)], &[])],
vec![hand_message(METHOD_RETURN, 13, &[serial_field(3)], &[])],
],
};
serve_bus(listener, script, sender);
let mut bus: Bus = Bus::connect(socket.as_path(), Duration::from_secs(10)).expect("connected");
bus.reload().expect("reloaded");
bus.start_unit("cli-git.socket").expect("started");
let (auth, calls) = receiver.recv().expect("seen");
let user: u32 = unsafe { libc::geteuid() };
let digits: String = user.to_string().bytes().map(|byte| { return format!("{byte:02x}"); }).collect();
assert_eq!(auth, format!("\0AUTH EXTERNAL {digits}\r\n"));
assert_eq!(calls, vec![encode_call(1, &HELLO, &[]), encode_call(2, &RELOAD, &[]), encode_call(3, &START_UNIT, &["cli-git.socket", "replace"])]);
remove(root.as_path());
}
fn serve_after_hello(listener: UnixListener, delay: Duration, answer: Option<Vec<u8>>, hold: Duration) {
std::thread::spawn(move || {
let (mut stream, _) = accept_and_authenticate(&listener);
let _hello = read_call(&mut stream);
stream.write_all(&string_return(1, ":1.42")).expect("the answer to hello");
let _call = read_call(&mut stream);
std::thread::sleep(delay);
if let Some(bytes) = answer {
let _written = stream.write_all(&bytes);
}
std::thread::sleep(hold);
});
}
#[test]
fn a_call_is_bounded_by_the_time_it_was_given_and_a_timeout_is_told_apart() {
let root = short_directory("within-bound");
let socket = root.join("bus");
let listener = UnixListener::bind(&socket).expect("listen");
let wait: Duration = Duration::from_millis(300);
let patience: Duration = Duration::from_secs(10);
serve_after_hello(listener, patience, None, Duration::from_secs(600));
let mut bus: Bus = Bus::connect(socket.as_path(), patience).expect("connected");
bus.within(wait);
assert_eq!(bus.reload(), Err(CallError::TimedOut));
remove(root.as_path());
}
#[test]
fn a_new_time_replaces_what_was_left_of_the_connections() {
let root = short_directory("within-extends");
let given: Duration = Duration::from_millis(200);
let spent: Duration = given * 2;
let patience: Duration = Duration::from_secs(10);
let control_socket = root.join("control");
serve_after_hello(UnixListener::bind(&control_socket).expect("listen"), Duration::ZERO, Some(hand_message(METHOD_RETURN, 12, &[serial_field(2)], &[])), Duration::from_secs(1));
let mut control: Bus = Bus::connect(control_socket.as_path(), given).expect("connected");
std::thread::sleep(spent);
assert_eq!(control.reload(), Err(CallError::TimedOut), "the connection's own time is spent");
let socket = root.join("bus");
serve_after_hello(UnixListener::bind(&socket).expect("listen"), Duration::ZERO, Some(hand_message(METHOD_RETURN, 12, &[serial_field(2)], &[])), Duration::from_secs(1));
let mut bus: Bus = Bus::connect(socket.as_path(), given).expect("connected");
std::thread::sleep(spent);
bus.within(patience);
assert_eq!(bus.reload(), Ok(()));
remove(root.as_path());
}
#[test]
fn an_error_reply_names_the_error() {
let root = short_directory("error");
let socket = root.join("bus");
let listener = UnixListener::bind(&socket).expect("listen");
let (sender, _receiver) = std::sync::mpsc::channel();
let mut body: Vec<u8> = Vec::new();
body.extend(9u32.to_le_bytes());
body.extend(b"not found\0");
let error: Vec<u8> = hand_message(
ERROR,
12,
&[hand_field(FIELD_ERROR_NAME, b's', "org.freedesktop.systemd1.NoSuchUnit"), serial_field(2), vec![FIELD_SIGNATURE, 1, b'g', 0, 1, b's', 0]],
&body,
);
serve_bus(listener, Script { answers: vec![vec![string_return(1, ":1.42")], vec![error]] }, sender);
let mut bus: Bus = Bus::connect(socket.as_path(), Duration::from_secs(10)).expect("connected");
assert_eq!(
bus.start_unit("x.socket"),
Err(CallError::Manager { name: String::from("org.freedesktop.systemd1.NoSuchUnit"), text: String::from("not found") })
);
remove(root.as_path());
}
#[test]
fn refusals_missing_sockets_and_silence_are_told_apart() {
let root = short_directory("refused");
let socket = root.join("bus");
let listener = UnixListener::bind(&socket).expect("listen");
std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("a client");
let mut one: [u8; 1] = [0];
let mut seen: Vec<u8> = Vec::new();
while !seen.ends_with(b"\r\n") {
stream.read_exact(&mut one).expect("bytes");
seen.push(one[0]);
}
stream.write_all(b"REJECTED EXTERNAL\r\n").expect("reject");
});
assert!(matches!(Bus::connect(socket.as_path(), Duration::from_secs(10)), Err(CallError::Authentication(_))));
assert!(matches!(Bus::connect(root.join("nobody").as_path(), Duration::from_secs(10)), Err(CallError::Connection(_))));
let silent = root.join("silent");
let quiet_listener = UnixListener::bind(&silent).expect("listen");
let (release, held) = std::sync::mpsc::channel::<()>();
let holder = std::thread::spawn(move || {
let (stream, _) = quiet_listener.accept().expect("a client");
let _released = held.recv();
drop(stream);
});
let wait: Duration = Duration::from_millis(300);
assert!(Bus::connect(silent.as_path(), wait).is_err());
release.send(()).expect("the holder lets go");
holder.join().expect("holder");
remove(root.as_path());
}
#[test]
#[ignore = "touches the real user service manager; run by hand"]
fn the_real_user_manager_loads_and_starts_a_runtime_only_probe_unit() {
let runtime = std::path::PathBuf::from(std::env::var_os("XDG_RUNTIME_DIR").expect("a runtime directory"));
let bus_path = session_bus_path(std::env::var("DBUS_SESSION_BUS_ADDRESS").ok().as_deref(), Some(runtime.as_path())).expect("a session bus");
let name = format!("cli-git-probe-{}", std::process::id());
let units = runtime.join("systemd").join("user");
std::fs::create_dir_all(&units).expect("unit directory");
let socket_path = runtime.join(format!("{name}.sock"));
std::fs::write(units.join(format!("{name}.socket")), format!("[Socket]\nListenStream={}\n", socket_path.display())).expect("socket unit");
std::fs::write(units.join(format!("{name}.service")), "[Service]\nExecStart=/usr/bin/true\n").expect("service unit");
let stop = Target { destination: MANAGER_NAME, path: MANAGER_PATH, interface: MANAGER_INTERFACE, member: "StopUnit" };
let outcome = (|| -> Result<(), CallError> {
let mut bus = Bus::connect(bus_path.as_path(), EXCHANGE_WAIT)?;
bus.start_unit(format!("{name}.socket").as_str())?;
let began = Instant::now();
let mut listening = UnixStream::connect(&socket_path).is_ok();
while !listening && began.elapsed() < Duration::from_secs(60) {
std::thread::sleep(Duration::from_millis(50));
listening = UnixStream::connect(&socket_path).is_ok();
}
let stopped = bus.call(&stop, &[format!("{name}.socket").as_str(), "replace"]);
assert!(listening, "the manager made the socket listen");
return stopped.map(drop);
})();
let _removed_socket = std::fs::remove_file(units.join(format!("{name}.socket")));
let _removed_service = std::fs::remove_file(units.join(format!("{name}.service")));
assert_eq!(outcome, Ok(()));
}