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

git_policy_cli/
systemd_dbus.rs
