git_policy_cli/
systemd_dbus.rs

1//! What:
2//!  A minimal D-Bus client for the user session bus: authenticate as the current user, say `Hello`, and call methods of
3//!  `org.freedesktop.systemd1.Manager` with string arguments, each bounded by one deadline for the whole exchange.
4//! Why:
5//!  The wrapper provisions socket activation itself, and may start no program to do it (`systemctl` would be one). The
6//!  supported surface of the user manager is its D-Bus API (`org.freedesktop.systemd1(5)`), reached through the session
7//!  bus; the manager's own private socket is documented as private to systemd. Only `Reload` and `StartUnit` are needed,
8//!  both with at most two string arguments, so the writer and the reader below cover exactly the types those calls and
9//!  the bus's replies use (`s`, `o`, `g`, `u`) and stop at anything else. The reader is total over arbitrary bytes (a
10//!  fuzz target holds it to that). The comparison with a D-Bus crate is in `doc/planning/cli-git-resident-service.md`,
11//!  "Provisioning".
12//!
13//! In TS you'd write (pseudocode):
14//! ```ts
15//! const bus = await connect(address, 5000); await bus.startUnit('cli-git.socket', 'replace'); // throws with the error name
16//! ```
17
18use std::io::{Read, Write};
19use std::os::unix::net::UnixStream;
20use std::path::Path;
21use std::time::{Duration, Instant};
22
23/// The manager's object path.
24pub const MANAGER_PATH: &str = "/org/freedesktop/systemd1";
25
26/// The manager's interface.
27pub const MANAGER_INTERFACE: &str = "org.freedesktop.systemd1.Manager";
28
29/// How long the whole exchange (connect, authenticate, say hello, make the calls) may take; a stuck bus never holds a
30///  commit longer.
31pub const EXCHANGE_WAIT: Duration = Duration::from_secs(20);
32
33/// The bus daemon's name, object and interface, for `Hello`.
34const BUS_NAME: &str = "org.freedesktop.DBus";
35const BUS_PATH: &str = "/org/freedesktop/DBus";
36
37/// The manager's bus name.
38pub const MANAGER_NAME: &str = "org.freedesktop.systemd1";
39
40/// The most bytes of one reply this client accepts.
41const MAX_REPLY_BYTES: usize = 1 << 20;
42
43/// Message type of a method call.
44const METHOD_CALL: u8 = 1;
45
46/// Message type of a successful reply.
47const METHOD_RETURN: u8 = 2;
48
49/// Message type of an error reply.
50const ERROR: u8 = 3;
51
52/// Header field code of the number of the call a reply answers.
53const FIELD_REPLY_SERIAL: u8 = 5;
54
55/// Header field codes.
56const FIELD_PATH: u8 = 1;
57const FIELD_INTERFACE: u8 = 2;
58const FIELD_MEMBER: u8 = 3;
59const FIELD_ERROR_NAME: u8 = 4;
60const FIELD_DESTINATION: u8 = 6;
61const FIELD_SIGNATURE: u8 = 8;
62
63/// Why a call did not succeed.
64#[derive(Clone, Debug, Eq, PartialEq)]
65pub enum CallError {
66    /// The socket could not be reached or the connection broke.
67    Connection(String),
68    /// The manager refused the authentication.
69    Authentication(String),
70    /// The manager answered with an error.
71    Manager {
72        /// The error's name, for example `org.freedesktop.systemd1.NoSuchUnit`.
73        name: String,
74        /// The error's text, when it gave one.
75        text: String,
76    },
77    /// The reply was not a message this client reads.
78    Malformed(String),
79}
80
81/// Appends values in D-Bus little-endian wire form.
82struct Writer {
83    bytes: Vec<u8>,
84}
85
86impl Writer {
87    fn align(&mut self, boundary: usize) {
88        while self.bytes.len() % boundary != 0 {
89            self.bytes.push(0);
90        }
91    }
92
93    fn unsigned(&mut self, value: u32) {
94        self.align(4);
95        self.bytes.extend(value.to_le_bytes());
96    }
97
98    fn text(&mut self, value: &str) {
99        self.unsigned(u32::try_from(value.len()).unwrap_or(u32::MAX));
100        self.bytes.extend(value.as_bytes());
101        self.bytes.push(0);
102    }
103
104    fn signature(&mut self, value: &str) {
105        self.bytes.push(u8::try_from(value.len()).unwrap_or(u8::MAX));
106        self.bytes.extend(value.as_bytes());
107        self.bytes.push(0);
108    }
109
110    /// One header field: its code, then a variant holding a string-like or signature value.
111    fn field(&mut self, code: u8, kind: char, value: &str) {
112        self.align(8);
113        self.bytes.push(code);
114        self.signature(&kind.to_string());
115        if kind == 'g' {
116            self.signature(value);
117        } else {
118            self.text(value);
119        }
120    }
121}
122
123/// Where a method call goes.
124#[derive(Clone, Copy, Debug, Eq, PartialEq)]
125pub struct Target {
126    /// The bus name of the receiver.
127    pub destination: &'static str,
128    /// The object path.
129    pub path: &'static str,
130    /// The interface.
131    pub interface: &'static str,
132    /// The method.
133    pub member: &'static str,
134}
135
136/// `Hello` on the bus daemon, the first call of every connection.
137pub const HELLO: Target = Target { destination: BUS_NAME, path: BUS_PATH, interface: BUS_NAME, member: "Hello" };
138
139/// `Reload` on the user manager.
140pub const RELOAD: Target = Target { destination: MANAGER_NAME, path: MANAGER_PATH, interface: MANAGER_INTERFACE, member: "Reload" };
141
142/// `StartUnit` on the user manager.
143pub const START_UNIT: Target = Target { destination: MANAGER_NAME, path: MANAGER_PATH, interface: MANAGER_INTERFACE, member: "StartUnit" };
144
145/// What:
146///  The bytes of a method call with string arguments.
147/// Why:
148///   Kept apart from the socket so a test can read them back and compare them with the specification's layout.
149///
150/// In TS you'd write (pseudocode):
151/// ```ts
152/// encodeCall(serial, START_UNIT, ['cli-git.socket', 'replace'])
153/// ```
154pub fn encode_call(serial: u32, target: &Target, arguments: &[&str]) -> Vec<u8> {
155    let mut body: Writer = Writer { bytes: Vec::new() };
156    for argument in arguments {
157        body.text(argument);
158    }
159    let mut fields: Writer = Writer { bytes: vec![0; 16] };
160    fields.field(FIELD_PATH, 'o', target.path);
161    fields.field(FIELD_DESTINATION, 's', target.destination);
162    fields.field(FIELD_INTERFACE, 's', target.interface);
163    fields.field(FIELD_MEMBER, 's', target.member);
164    if !arguments.is_empty() {
165        fields.field(FIELD_SIGNATURE, 'g', &"s".repeat(arguments.len()));
166    }
167    // The first sixteen bytes are the fixed header and the array length, filled in below.
168    let array_length: u32 = u32::try_from(fields.bytes.len() - 16).unwrap_or(u32::MAX);
169    fields.align(8);
170    let mut message: Vec<u8> = fields.bytes;
171    message[0] = b'l';
172    message[1] = METHOD_CALL;
173    message[2] = 0;
174    message[3] = 1;
175    message[4..8].copy_from_slice(&u32::try_from(body.bytes.len()).unwrap_or(u32::MAX).to_le_bytes());
176    message[8..12].copy_from_slice(&serial.to_le_bytes());
177    message[12..16].copy_from_slice(&array_length.to_le_bytes());
178    message.extend(body.bytes);
179    return message;
180}
181
182/// A reply: its type, the error name its header carries, and the first string of its body.
183#[derive(Clone, Debug, Eq, PartialEq)]
184pub struct Reply {
185    /// Message type: 2 for a return, 3 for an error, 4 for a signal.
186    pub kind: u8,
187    /// The number of the call this message answers.
188    pub reply_serial: Option<u32>,
189    /// The error name of an error reply.
190    pub error_name: Option<String>,
191    /// The first string argument of the body, which an error reply uses for its text.
192    pub first_text: Option<String>,
193}
194
195fn read_u32(bytes: &[u8], at: usize) -> Option<u32> {
196    return bytes.get(at..at.checked_add(4)?).map(|slice| { return u32::from_le_bytes([slice[0], slice[1], slice[2], slice[3]]); });
197}
198
199fn read_text(bytes: &[u8], at: usize) -> Option<(String, usize)> {
200    let length: usize = usize::try_from(read_u32(bytes, at)?).ok()?;
201    let start: usize = at.checked_add(4)?;
202    let text: &[u8] = bytes.get(start..start.checked_add(length)?)?;
203    return Some((String::from_utf8_lossy(text).into_owned(), start.saturating_add(length).saturating_add(1)));
204}
205
206fn aligned(offset: usize, boundary: usize) -> usize {
207    return offset.checked_add(boundary - 1).map_or(usize::MAX, |sum| { return sum / boundary * boundary; });
208}
209
210/// What:
211///  Reads a complete reply message.
212/// Why:
213///   Only the type, the error name and the first string of the body matter; a header field of a kind this reader does not
214///  know ends the walk of the header, which is safe because the fields it needs come first in what the manager sends or
215///  are looked for before any unknown kind.
216///
217/// In TS you'd write (pseudocode):
218/// ```ts
219/// decodeReply(bytes)
220/// ```
221pub fn decode_reply(bytes: &[u8]) -> Result<Reply, CallError> {
222    let malformed = |what: &str| { return CallError::Malformed(what.to_string()); };
223    if bytes.len() < 16 || bytes[0] != b'l' {
224        return Err(malformed("the reply is not a little-endian message"));
225    }
226    let kind: u8 = bytes[1];
227    let body_length: usize = usize::try_from(read_u32(bytes, 4).ok_or_else(|| { return malformed("no body length"); })?).unwrap_or(usize::MAX);
228    let fields_length: usize = usize::try_from(read_u32(bytes, 12).ok_or_else(|| { return malformed("no header length"); })?).unwrap_or(usize::MAX);
229    let fields_end: usize = 16usize.saturating_add(fields_length);
230    let body_start: usize = aligned(fields_end, 8);
231    if bytes.len() < body_start.saturating_add(body_length) || fields_end > bytes.len() {
232        return Err(malformed("the reply is shorter than it says"));
233    }
234    let mut error_name: Option<String> = None;
235    let mut reply_serial: Option<u32> = None;
236    let mut signature: String = String::new();
237    let mut at: usize = 16;
238    while at < fields_end {
239        at = aligned(at, 8);
240        let (Some(&code), Some(&signature_length)) = (bytes.get(at), bytes.get(at.saturating_add(1))) else { break };
241        let kind_at: usize = at.saturating_add(2);
242        let Some(&value_kind) = bytes.get(kind_at) else { break };
243        let value_at: usize = kind_at.saturating_add(usize::from(signature_length)).saturating_add(1);
244        match (value_kind, code) {
245            (b's' | b'o', _) => {
246                let Some((text, next)) = read_text(bytes, aligned(value_at, 4)) else { break };
247                if code == FIELD_ERROR_NAME {
248                    error_name = Some(text);
249                }
250                at = next;
251            }
252            (b'g', _) => {
253                let length: usize = usize::from(*bytes.get(value_at).unwrap_or(&0));
254                if code == FIELD_SIGNATURE {
255                    let start: usize = value_at.saturating_add(1);
256                    signature = String::from_utf8_lossy(bytes.get(start..start.saturating_add(length)).unwrap_or_default()).into_owned();
257                }
258                at = value_at.saturating_add(length).saturating_add(2);
259            }
260            (b'u', _) => {
261                let number_at: usize = aligned(value_at, 4);
262                if code == FIELD_REPLY_SERIAL {
263                    reply_serial = read_u32(bytes, number_at);
264                }
265                at = number_at.saturating_add(4);
266            }
267            _ => break,
268        }
269    }
270    let first_text: Option<String> = if signature.starts_with('s') {
271        read_text(bytes, aligned(body_start, 4)).map(|(text, _)| { return text; })
272    } else {
273        None
274    };
275    return Ok(Reply { kind, reply_serial, error_name, first_text });
276}
277
278/// The user id as the ASCII digits the `EXTERNAL` mechanism sends, written in hexadecimal.
279fn external_identity() -> String {
280    // SAFETY: `geteuid` takes no arguments and cannot fail.
281    let user: u32 = unsafe { libc::geteuid() };
282    return user.to_string().bytes().map(|byte| { return format!("{byte:02x}"); }).collect();
283}
284
285/// The time left of the exchange, or an error when it ran out.
286fn remaining(deadline: Instant) -> Result<Duration, CallError> {
287    let left: Duration = deadline.saturating_duration_since(Instant::now());
288    return if left.is_zero() { Err(CallError::Connection(String::from("the bus did not answer in time"))) } else { Ok(left) };
289}
290
291fn read_line(stream: &mut UnixStream, deadline: Instant) -> Result<String, CallError> {
292    let mut line: Vec<u8> = Vec::new();
293    let mut one: [u8; 1] = [0];
294    while line.len() < 4096 {
295        stream.set_read_timeout(Some(remaining(deadline)?)).map_err(|error| { return CallError::Connection(error.to_string()); })?;
296        stream.read_exact(&mut one).map_err(|error| { return CallError::Authentication(error.to_string()); })?;
297        if one[0] == b'\n' {
298            return Ok(String::from_utf8_lossy(line.as_slice()).trim_end_matches('\r').to_string());
299        }
300        line.push(one[0]);
301    }
302    return Err(CallError::Authentication(String::from("the answer is too long")));
303}
304
305/// Reads exactly one message from the stream.
306fn read_message(stream: &mut UnixStream, deadline: Instant) -> Result<Vec<u8>, CallError> {
307    let mut header: Vec<u8> = vec![0; 16];
308    stream.set_read_timeout(Some(remaining(deadline)?)).map_err(|error| { return CallError::Connection(error.to_string()); })?;
309    stream.read_exact(header.as_mut_slice()).map_err(|error| { return CallError::Connection(error.to_string()); })?;
310    let body_length: usize = usize::try_from(read_u32(header.as_slice(), 4).unwrap_or(0)).unwrap_or(usize::MAX);
311    let fields_length: usize = usize::try_from(read_u32(header.as_slice(), 12).unwrap_or(0)).unwrap_or(usize::MAX);
312    let rest: usize = aligned(16usize.saturating_add(fields_length), 8).saturating_add(body_length).saturating_sub(16);
313    if rest > MAX_REPLY_BYTES {
314        return Err(CallError::Malformed(String::from("the reply is larger than this client reads")));
315    }
316    let mut remainder: Vec<u8> = vec![0; rest];
317    stream.set_read_timeout(Some(remaining(deadline)?)).map_err(|error| { return CallError::Connection(error.to_string()); })?;
318    stream.read_exact(remainder.as_mut_slice()).map_err(|error| { return CallError::Connection(error.to_string()); })?;
319    header.extend(remainder);
320    return Ok(header);
321}
322
323/// The path of the session bus socket: the `unix:path=` entry of the address variable, else `$XDG_RUNTIME_DIR/bus`.
324pub fn session_bus_path(address: Option<&str>, runtime_dir: Option<&Path>) -> Option<std::path::PathBuf> {
325    let from_address: Option<std::path::PathBuf> = address.and_then(|list| {
326        return list.split(';').find_map(|entry| {
327            let rest: &str = entry.strip_prefix("unix:")?;
328            let path: &str = rest.split(',').find_map(|option| { return option.strip_prefix("path="); })?;
329            // Escapes (`%xx`) in a path are not read; a path that uses them falls back to the default location.
330            return if path.contains('%') || path.is_empty() { None } else { Some(std::path::PathBuf::from(path)) };
331        });
332    });
333    return from_address.or_else(|| { return runtime_dir.map(|directory| { return directory.join("bus"); }); });
334}
335
336/// An authenticated connection to the session bus, past `Hello`.
337pub struct Bus {
338    stream: UnixStream,
339    deadline: Instant,
340    serial: u32,
341}
342
343/// Methods over a connection.
344impl Bus {
345    /// What:
346    ///  Connects to the bus socket, authenticates as the current user and says `Hello`, all within `wait`.
347    /// Why:
348    ///   The deadline of the whole exchange starts here, so every later call shares what is left of it.
349    ///
350    /// In TS you'd write (pseudocode):
351    /// ```ts
352    /// const bus = await Bus.connect(path, 20000)
353    /// ```
354    pub fn connect(socket: &Path, wait: Duration) -> Result<Bus, CallError> {
355        let deadline: Instant = Instant::now() + wait;
356        let mut stream: UnixStream = UnixStream::connect(socket).map_err(|error| { return CallError::Connection(error.to_string()); })?;
357        stream.set_write_timeout(Some(wait)).map_err(|error| { return CallError::Connection(error.to_string()); })?;
358        let mut hello: Vec<u8> = vec![0];
359        hello.extend(format!("AUTH EXTERNAL {}\r\n", external_identity()).as_bytes());
360        stream.write_all(hello.as_slice()).map_err(|error| { return CallError::Authentication(error.to_string()); })?;
361        let answer: String = read_line(&mut stream, deadline)?;
362        if !answer.starts_with("OK ") {
363            return Err(CallError::Authentication(answer));
364        }
365        stream.write_all(b"BEGIN\r\n").map_err(|error| { return CallError::Authentication(error.to_string()); })?;
366        let mut bus: Bus = Bus { stream, deadline, serial: 0 };
367        bus.call(&HELLO, &[])?;
368        return Ok(bus);
369    }
370
371    /// What:
372    ///  Makes one method call and waits for the reply that answers it, skipping the signals the bus sends meanwhile.
373    /// Why:
374    ///   A return is the receiver's confirmation (a reload returns after the units are read); an error reply comes back
375    ///  with its name and text so the caller can say what to do.
376    ///
377    /// In TS you'd write (pseudocode):
378    /// ```ts
379    /// await bus.call(RELOAD, [])
380    /// ```
381    pub fn call(&mut self, target: &Target, arguments: &[&str]) -> Result<Reply, CallError> {
382        self.serial += 1;
383        let serial: u32 = self.serial;
384        self.stream.write_all(encode_call(serial, target, arguments).as_slice()).map_err(|error| { return CallError::Connection(error.to_string()); })?;
385        loop {
386            let reply: Reply = decode_reply(read_message(&mut self.stream, self.deadline)?.as_slice())?;
387            if reply.reply_serial != Some(serial) || !matches!(reply.kind, METHOD_RETURN | ERROR) {
388                continue;
389            }
390            return match (reply.kind, reply.error_name.clone()) {
391                (METHOD_RETURN, _) => Ok(reply),
392                (_, Some(name)) => Err(CallError::Manager { name, text: reply.first_text.unwrap_or_default() }),
393                _ => Err(CallError::Malformed(String::from("an error reply without a name"))),
394            };
395        }
396    }
397
398    /// Reloads the user manager's unit files.
399    pub fn reload(&mut self) -> Result<(), CallError> {
400        return self.call(&RELOAD, &[]).map(drop);
401    }
402
403    /// Starts a unit, replacing a queued job for it.
404    pub fn start_unit(&mut self, name: &str) -> Result<(), CallError> {
405        return self.call(&START_UNIT, &[name, "replace"]).map(drop);
406    }
407}
408
409/// Client controls stay out of the release executable.
410#[cfg(test)]
411#[path = "systemd_dbus_tests.rs"]
412mod tests;
git_policy_cli/systemd_dbus.rs

git_policy_cli/
systemd_dbus.rs
