✨ zb: Add support for unixexec transport

zbus/Cargo.toml

@@ -110,7 +110,7 @@ nix = { version = "0.29", default-features = false, features = [
   "user",
 ] }
 
-[target.'cfg(target_os = "macos")'.dependencies]
+[target.'cfg(target_family = "unix")'.dependencies]
 # FIXME: This should only be enabled if async-io feature is enabled but currently
 # Cargo doesn't provide a way to do that for only specific target OS: https://github.com/rust-lang/cargo/issues/1197.
 async-process = "2.2.2"

zbus/src/abstractions/mod.rs

@@ -8,6 +8,6 @@ mod async_drop;
 pub(crate) mod async_lock;
 pub use async_drop::*;
 
-// Not macOS-specific itself but only used on macOS.
-#[cfg(target_os = "macos")]
+// Not unix-specific itself but only used on unix.
+#[cfg(target_family = "unix")]
 pub(crate) mod process;

zbus/src/abstractions/process.rs

@@ -1,20 +1,103 @@
-use std::{ffi::OsStr, io::Error, process::Output};
+#[cfg(not(feature = "tokio"))]
+use async_process::{unix::CommandExt, Child};
+#[cfg(target_os = "macos")]
+use std::process::Output;
+use std::{ffi::OsStr, io::Error, process::Stdio};
+#[cfg(feature = "tokio")]
+use tokio::process::Child;
+
+use crate::address::transport::Unixexec;
+
+/// A wrapper around the command API of the underlying async runtime.
+pub struct Command(
+    #[cfg(not(feature = "tokio"))] async_process::Command,
+    #[cfg(feature = "tokio")] tokio::process::Command,
+);
+
+impl Command {
+    /// Constructs a new `Command` for launching the program at path `program`.
+    pub fn new<S>(program: S) -> Self
+    where
+        S: AsRef<OsStr>,
+    {
+        #[cfg(not(feature = "tokio"))]
+        return Self(async_process::Command::new(program));
+
+        #[cfg(feature = "tokio")]
+        return Self(tokio::process::Command::new(program));
+    }
+
+    /// Constructs a new `Command` from a `unixexec` address.
+    pub fn for_unixexec(unixexec: &Unixexec) -> Self {
+        let mut command = Self::new(unixexec.path());
+        command.args(unixexec.args());
+
+        if let Some(arg0) = unixexec.arg0() {
+            command.arg0(arg0);
+        }
+
+        command
+    }
+
+    /// Sets executable argument.
+    ///
+    /// Set the first process argument, `argv[0]`, to something other than the
+    /// default executable path.
+    pub fn arg0<S>(&mut self, arg: S) -> &mut Self
+    where
+        S: AsRef<OsStr>,
+    {
+        self.0.arg0(arg);
+        self
+    }
+
+    /// Adds multiple arguments to pass to the program.
+    pub fn args<I, S>(&mut self, args: I) -> &mut Self
+    where
+        I: IntoIterator<Item = S>,
+        S: AsRef<OsStr>,
+    {
+        self.0.args(args);
+        self
+    }
+
+    /// Executes the command as a child process, waiting for it to finish and
+    /// collecting all of its output.
+    #[cfg(target_os = "macos")]
+    pub async fn output(&mut self) -> Result<Output, Error> {
+        self.0.output().await
+    }
+
+    /// Sets configuration for the child process's standard input (stdin) handle.
+    pub fn stdin<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.0.stdin(cfg);
+        self
+    }
+
+    /// Sets configuration for the child process's standard output (stdout) handle.
+    pub fn stdout<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.0.stdout(cfg);
+        self
+    }
+
+    /// Sets configuration for the child process's standard error (stderr) handle.
+    pub fn stderr<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.0.stderr(cfg);
+        self
+    }
+
+    /// Executes the command as a child process, returning a handle to it.
+    pub fn spawn(&mut self) -> Result<Child, Error> {
+        self.0.spawn()
+    }
+}
 
 /// An asynchronous wrapper around running and getting command output
+#[cfg(target_os = "macos")]
 pub async fn run<I, S>(program: S, args: I) -> Result<Output, Error>
 where
     I: IntoIterator<Item = S>,
     S: AsRef<OsStr>,
 {
-    #[cfg(not(feature = "tokio"))]
-    return async_process::Command::new(program)
-        .args(args)
-        .output()
-        .await;
-
-    #[cfg(feature = "tokio")]
-    return tokio::process::Command::new(program)
-        .args(args)
-        .output()
-        .await;
+    Command::new(program).args(args).output().await
 }

zbus/src/address/mod.rs

@@ -182,6 +182,8 @@ mod tests {
     };
     #[cfg(target_os = "macos")]
     use crate::address::transport::Launchd;
+    #[cfg(unix)]
+    use crate::address::transport::Unixexec;
     #[cfg(windows)]
     use crate::address::transport::{Autolaunch, AutolaunchScope};
     use crate::{
@@ -232,6 +234,11 @@ mod tests {
             }
             _ => panic!(),
         }
+        #[cfg(unix)]
+        match Address::from_str("unixexec:foo=blah").unwrap_err() {
+            Error::Address(e) => assert_eq!(e, "unixexec address is missing `path`"),
+            _ => panic!(),
+        }
         assert_eq!(
             Address::from_str("unix:path=/tmp/dbus-foo").unwrap(),
             Transport::Unix(Unix::new(UnixSocket::File("/tmp/dbus-foo".into()))).into(),
@@ -253,6 +260,11 @@ mod tests {
                 .unwrap(),
             );
         }
+        #[cfg(unix)]
+        assert_eq!(
+            Address::from_str("unixexec:path=/tmp/dbus-foo").unwrap(),
+            Transport::Unixexec(Unixexec::new("/tmp/dbus-foo".into(), None, Vec::new())).into(),
+        );
         assert_eq!(
             Address::from_str("tcp:host=localhost,port=4142").unwrap(),
             Transport::Tcp(Tcp::new("localhost", 4142)).into(),

zbus/src/address/transport/mod.rs

@@ -1,7 +1,9 @@
 //! D-Bus transport Information module.
 //!
-//! This module provides the trasport information for D-Bus addresses.
+//! This module provides the transport information for D-Bus addresses.
 
+#[cfg(unix)]
+use crate::connection::socket::Command;
 #[cfg(windows)]
 use crate::win32::autolaunch_bus_address;
 use crate::{Error, Result};
@@ -20,6 +22,10 @@ use tokio_vsock::VsockStream;
 use uds_windows::UnixStream;
 #[cfg(all(feature = "vsock", not(feature = "tokio")))]
 use vsock::VsockStream;
+#[cfg(unix)]
+mod unixexec;
+#[cfg(unix)]
+pub use unixexec::Unixexec;
 
 use std::{
     fmt::{Display, Formatter},
@@ -77,6 +83,9 @@ pub enum Transport {
     /// The type of `stream` is `vsock::VsockStream` with the `vsock` feature and
     /// `tokio_vsock::VsockStream` with the `tokio-vsock` feature.
     Vsock(Vsock),
+    /// A `unixexec` address.
+    #[cfg(unix)]
+    Unixexec(Unixexec),
 }
 
 impl Transport {
@@ -136,6 +145,8 @@ impl Transport {
                     }
                 }
             }
+            #[cfg(unix)]
+            Transport::Unixexec(unixexec) => unixexec.connect().await.map(Stream::Unixexec),
             #[cfg(all(feature = "vsock", not(feature = "tokio")))]
             Transport::Vsock(addr) => {
                 let stream = VsockStream::connect_with_cid_port(addr.cid(), addr.port())?;
@@ -213,6 +224,8 @@ impl Transport {
     pub(super) fn from_options(transport: &str, options: HashMap<&str, &str>) -> Result<Self> {
         match transport {
             "unix" => Unix::from_options(options).map(Self::Unix),
+            #[cfg(unix)]
+            "unixexec" => Unixexec::from_options(options).map(Self::Unixexec),
             "tcp" => Tcp::from_options(options, false).map(Self::Tcp),
             "nonce-tcp" => Tcp::from_options(options, true).map(Self::Tcp),
             #[cfg(any(
@@ -236,6 +249,8 @@ impl Transport {
 #[derive(Debug)]
 pub(crate) enum Stream {
     Unix(Async<UnixStream>),
+    #[cfg(unix)]
+    Unixexec(Command),
     Tcp(Async<TcpStream>),
     #[cfg(feature = "vsock")]
     Vsock(Async<VsockStream>),
@@ -246,6 +261,8 @@ pub(crate) enum Stream {
 pub(crate) enum Stream {
     #[cfg(unix)]
     Unix(tokio::net::UnixStream),
+    #[cfg(unix)]
+    Unixexec(Command),
     Tcp(TcpStream),
     #[cfg(feature = "tokio-vsock")]
     Vsock(VsockStream),
@@ -336,6 +353,8 @@ impl Display for Transport {
         match self {
             Self::Tcp(tcp) => write!(f, "{}", tcp)?,
             Self::Unix(unix) => write!(f, "{}", unix)?,
+            #[cfg(unix)]
+            Self::Unixexec(unixexec) => write!(f, "{}", unixexec)?,
             #[cfg(any(
                 all(feature = "vsock", not(feature = "tokio")),
                 feature = "tokio-vsock"

zbus/src/address/transport/unixexec.rs

@@ -0,0 +1,115 @@
+use std::{
+    borrow::BorrowMut, ffi::OsString, fmt::Display, os::unix::ffi::OsStrExt, path::PathBuf,
+    process::Stdio,
+};
+
+use crate::process::Command;
+
+use super::encode_percents;
+
+/// `unixexec:` D-Bus transport.
+///
+/// <https://dbus.freedesktop.org/doc/dbus-specification.html#transports-exec>
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Unixexec {
+    path: PathBuf,
+    arg0: Option<OsString>,
+    args: Vec<OsString>,
+}
+
+impl Unixexec {
+    /// Create a new unixexec transport with the given path and arguments.
+    pub fn new(path: PathBuf, arg0: Option<OsString>, args: Vec<OsString>) -> Self {
+        Self { path, arg0, args }
+    }
+
+    pub(super) fn from_options(opts: std::collections::HashMap<&str, &str>) -> crate::Result<Self> {
+        let Some(path) = opts.get("path") else {
+            return Err(crate::Error::Address(
+                "unixexec address is missing `path`".to_owned(),
+            ));
+        };
+
+        let arg0 = opts.get("argv0").map(OsString::from);
+
+        let mut args: Vec<OsString> = Vec::new();
+        let mut arg_index = 1;
+        while let Some(arg) = opts.get(format!("argv{arg_index}").as_str()) {
+            args.push(OsString::from(arg));
+            arg_index += 1;
+        }
+
+        Ok(Self::new(PathBuf::from(path), arg0, args))
+    }
+
+    /// Binary to execute.
+    ///
+    /// Path of the binary to execute, either an absolute path or a binary name that is searched for
+    /// in the default search path of the OS. This corresponds to the first argument of execlp().
+    /// This key is mandatory.
+    pub fn path(&self) -> &PathBuf {
+        &self.path
+    }
+
+    /// The executable argument.
+    ///
+    /// The program name to use when executing the binary. If omitted the same
+    /// value as specified for path will be used. This corresponds to the
+    /// second argument of execlp().
+    pub fn arg0(&self) -> Option<&OsString> {
+        self.arg0.as_ref()
+    }
+
+    /// Arguments.
+    ///
+    /// Arguments to pass to the binary.
+    pub fn args(&self) -> &[OsString] {
+        self.args.as_ref()
+    }
+
+    pub(super) async fn connect(&self) -> crate::Result<crate::connection::socket::Command> {
+        Command::for_unixexec(self)
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit())
+            .spawn()?
+            .borrow_mut()
+            .try_into()
+    }
+}
+
+impl Display for Unixexec {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("unixexec:path=")?;
+        encode_percents(f, self.path.as_os_str().as_bytes())?;
+
+        if let Some(arg0) = self.arg0.as_ref() {
+            f.write_str(",argv0=")?;
+            encode_percents(f, arg0.as_bytes())?;
+        }
+
+        for (index, arg) in self.args.iter().enumerate() {
+            write!(f, ",argv{}=", index + 1)?;
+            encode_percents(f, arg.as_bytes())?;
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::address::{transport::Transport, Address};
+
+    #[test]
+    fn connect() {
+        let addr: Address = "unixexec:path=echo,argv1=hello,argv2=world"
+            .try_into()
+            .unwrap();
+        let unixexec = match addr.transport() {
+            Transport::Unixexec(unixexec) => unixexec,
+            _ => unreachable!(),
+        };
+        crate::utils::block_on(unixexec.connect()).unwrap();
+    }
+}

zbus/src/connection/builder.rs

@@ -492,6 +492,8 @@ impl<'a> Builder<'a> {
                 match address.connect().await? {
                     #[cfg(any(unix, not(feature = "tokio")))]
                     address::transport::Stream::Unix(stream) => stream.into(),
+                    #[cfg(unix)]
+                    address::transport::Stream::Unixexec(stream) => stream.into(),
                     address::transport::Stream::Tcp(stream) => stream.into(),
                     #[cfg(any(
                         all(feature = "vsock", not(feature = "tokio")),