zbus/address/transport/

mod.rs

//! D-Bus transport Information module.
//!
//! This module provides the transport information for D-Bus addresses.

#[cfg(windows)]
use crate::win32::autolaunch_bus_address;
use crate::{Address, Error, Result, connection::socket::BoxedSplit};
#[cfg(feature = "async-io")]
use async_io::Async;
#[cfg(unix)]
use std::os::unix::net::{SocketAddr, UnixStream};
use std::{collections::HashMap, sync::Arc};
#[cfg(windows)]
use uds_windows::UnixStream;
#[cfg(unix)]
mod unixexec;
#[cfg(unix)]
pub use unixexec::Unixexec;

use std::{
    fmt::{Display, Formatter},
    str::from_utf8_unchecked,
};

mod unix;
pub use unix::{Unix, UnixSocket};
mod tcp;
pub use tcp::{Tcp, TcpTransportFamily};
#[cfg(windows)]
mod autolaunch;
#[cfg(windows)]
pub use autolaunch::{Autolaunch, AutolaunchScope};
#[cfg(target_os = "macos")]
mod launchd;
#[cfg(target_os = "macos")]
pub use launchd::Launchd;
#[cfg(unix)]
mod ibus;
#[cfg(unix)]
pub use ibus::Ibus;
#[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
#[path = "vsock.rs"]
// Gotta rename to avoid name conflict with the `vsock` crate.
mod vsock_transport;
#[cfg(target_os = "linux")]
use std::os::linux::net::SocketAddrExt;
#[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
pub use vsock_transport::Vsock;

/// The transport properties of a D-Bus address.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum Transport {
    /// A Unix Domain Socket address.
    Unix(Unix),
    /// A TCP address.
    Tcp(Tcp),
    /// An autolaunch D-Bus address.
    #[cfg(windows)]
    Autolaunch(Autolaunch),
    /// A launchd D-Bus address.
    #[cfg(target_os = "macos")]
    Launchd(Launchd),
    /// An IBus D-Bus address.
    ///
    /// IBus (Intelligent Input Bus) is an input method framework. This transport queries the
    /// IBus daemon for its D-Bus address using the `ibus address` command.
    #[cfg(unix)]
    Ibus(Ibus),
    #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
    /// A VSOCK address.
    ///
    /// This variant is only available when either the `vsock` or `tokio-vsock` feature is enabled.
    /// 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 {
    #[cfg_attr(any(unix, windows), async_recursion::async_recursion)]
    pub(super) async fn connect(self, address: Address) -> Result<Stream> {
        match self {
            Transport::Unix(unix) => {
                // This is a `path` in case of Windows until uds_windows provides the needed API:
                // https://github.com/haraldh/rust_uds_windows/issues/14
                let addr = match unix.take_path() {
                    #[cfg(unix)]
                    UnixSocket::File(path) => SocketAddr::from_pathname(path)?,
                    #[cfg(windows)]
                    UnixSocket::File(path) => path,
                    #[cfg(target_os = "linux")]
                    UnixSocket::Abstract(name) => {
                        SocketAddr::from_abstract_name(name.as_encoded_bytes())?
                    }
                    UnixSocket::Dir(_) | UnixSocket::TmpDir(_) => {
                        // you can't connect to a unix:dir
                        return Err(Error::Unsupported);
                    }
                };
                let stream = crate::Task::spawn_blocking(
                    move || -> Result<_> {
                        #[cfg(unix)]
                        let stream = UnixStream::connect_addr(&addr)
                            .map_err(|e| Error::Connection(Arc::new(e), address))?;
                        #[cfg(windows)]
                        let stream = UnixStream::connect(addr)
                            .map_err(|e| Error::Connection(Arc::new(e), address))?;
                        stream.set_nonblocking(true)?;

                        Ok(stream)
                    },
                    "unix stream connection",
                )
                .await??;
                #[cfg(unix)]
                {
                    let split = crate::abstractions::select_runtime! {
                        tokio: unix_stream_to_tokio(stream),
                        async_io: unix_stream_to_async_io(stream),
                    };
                    split.map(Stream::Unix)
                }
                // tokio doesn't support unix sockets on Windows, so async-io is the only backend
                // that can drive one there.
                #[cfg(all(not(unix), feature = "async-io"))]
                {
                    unix_stream_to_async_io(stream).map(Stream::Unix)
                }
                #[cfg(all(not(unix), not(feature = "async-io")))]
                {
                    let _ = stream;
                    Err(Error::Unsupported)
                }
            }
            #[cfg(unix)]
            Transport::Unixexec(unixexec) => unixexec
                .connect(&address)
                .await
                .map(|s| Stream::Unixexec(s.into())),
            #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
            Transport::Vsock(addr) => {
                #[cfg(all(feature = "vsock", feature = "tokio-vsock"))]
                {
                    if crate::abstractions::use_tokio() {
                        vsock_connect_tokio(&addr, &address).await
                    } else {
                        vsock_connect_async_io(&addr, &address)
                    }
                }
                #[cfg(all(feature = "vsock", not(feature = "tokio-vsock")))]
                {
                    vsock_connect_async_io(&addr, &address)
                }
                #[cfg(all(feature = "tokio-vsock", not(feature = "vsock")))]
                {
                    vsock_connect_tokio(&addr, &address).await
                }
            }

            Transport::Tcp(mut addr) => {
                let nonce_file = addr.take_nonce_file();
                #[allow(unused_mut)]
                let mut stream = addr.connect(&address).await?;

                if let Some(nonce_file) = nonce_file {
                    #[cfg(unix)]
                    let nonce_file = {
                        use std::os::unix::ffi::OsStrExt;
                        std::ffi::OsStr::from_bytes(&nonce_file)
                    };

                    #[cfg(windows)]
                    let nonce_file = std::str::from_utf8(&nonce_file).map_err(|_| {
                        Error::Address("nonce file path is invalid UTF-8".to_owned())
                    })?;

                    #[cfg(feature = "async-io")]
                    {
                        let nonce = std::fs::read(nonce_file)?;
                        let mut nonce = &nonce[..];

                        while !nonce.is_empty() {
                            let len = stream
                                .write_with(|mut s| std::io::Write::write(&mut s, nonce))
                                .await?;
                            nonce = &nonce[len..];
                        }
                    }

                    #[cfg(all(feature = "tokio", not(feature = "async-io")))]
                    {
                        let nonce = tokio::fs::read(nonce_file).await?;
                        tokio::io::AsyncWriteExt::write_all(&mut stream, &nonce).await?;
                    }
                }

                #[cfg(feature = "async-io")]
                let split = tcp_async_to_split(stream)?;
                #[cfg(all(feature = "tokio", not(feature = "async-io")))]
                let split = stream.into();

                Ok(Stream::Tcp(split))
            }

            #[cfg(windows)]
            Transport::Autolaunch(Autolaunch { scope }) => match scope {
                Some(_) => Err(Error::Address(
                    "Autolaunch scopes are currently unsupported".to_owned(),
                )),
                None => {
                    let addr = autolaunch_bus_address()?;
                    addr.connect().await
                }
            },

            #[cfg(target_os = "macos")]
            Transport::Launchd(launchd) => {
                let transport = launchd.bus_address().await?;
                transport.connect(address).await
            }

            #[cfg(unix)]
            Transport::Ibus(ibus) => {
                let addr = ibus.bus_address().await?;
                addr.connect().await
            }
        }
    }

    // Helper for `FromStr` impl of `Address`.
    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(feature = "vsock", feature = "tokio-vsock"))]
            "vsock" => Vsock::from_options(options).map(Self::Vsock),
            #[cfg(windows)]
            "autolaunch" => Autolaunch::from_options(options).map(Self::Autolaunch),
            #[cfg(target_os = "macos")]
            "launchd" => Launchd::from_options(options).map(Self::Launchd),
            #[cfg(unix)]
            "ibus" => Ibus::from_options(options).map(Self::Ibus),

            _ => Err(Error::Address(format!(
                "unsupported transport '{transport}'"
            ))),
        }
    }
}

#[derive(Debug)]
pub(crate) enum Stream {
    #[cfg(any(unix, feature = "async-io"))]
    Unix(BoxedSplit),
    #[cfg(unix)]
    Unixexec(BoxedSplit),
    Tcp(BoxedSplit),
    #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
    Vsock(BoxedSplit),
}

#[cfg(feature = "async-io")]
fn unix_stream_to_async_io(stream: UnixStream) -> Result<BoxedSplit> {
    Async::new(stream).map(Into::into).map_err(Into::into)
}

#[cfg(all(unix, feature = "tokio"))]
fn unix_stream_to_tokio(stream: UnixStream) -> Result<BoxedSplit> {
    tokio::net::UnixStream::from_std(stream)
        .map(Into::into)
        .map_err(Into::into)
}

// `async-io` always does the connecting; hand the socket over to tokio when it's in use.
#[cfg(feature = "async-io")]
fn tcp_async_to_split(stream: Async<std::net::TcpStream>) -> Result<BoxedSplit> {
    #[cfg(feature = "tokio")]
    if crate::abstractions::use_tokio() {
        return tokio::net::TcpStream::from_std(stream.into_inner()?)
            .map(Into::into)
            .map_err(Into::into);
    }

    Ok(stream.into())
}

#[cfg(feature = "vsock")]
fn vsock_connect_async_io(addr: &Vsock, address: &Address) -> Result<Stream> {
    let stream = vsock::VsockStream::connect_with_cid_port(addr.cid(), addr.port())
        .map_err(|e| Error::Connection(Arc::new(e), address.clone()))?;
    Async::new(stream)
        .map(|s| Stream::Vsock(s.into()))
        .map_err(Into::into)
}

#[cfg(feature = "tokio-vsock")]
async fn vsock_connect_tokio(addr: &Vsock, address: &Address) -> Result<Stream> {
    tokio_vsock::VsockStream::connect(tokio_vsock::VsockAddr::new(addr.cid(), addr.port()))
        .await
        .map(|s| Stream::Vsock(s.into()))
        .map_err(|e| Error::Connection(Arc::new(e), address.clone()))
}

fn decode_hex(c: char) -> Result<u8> {
    match c {
        '0'..='9' => Ok(c as u8 - b'0'),
        'a'..='f' => Ok(c as u8 - b'a' + 10),
        'A'..='F' => Ok(c as u8 - b'A' + 10),

        _ => Err(Error::Address(
            "invalid hexadecimal character in percent-encoded sequence".to_owned(),
        )),
    }
}

pub(crate) fn decode_percents(value: &str) -> Result<Vec<u8>> {
    let mut iter = value.chars();
    let mut decoded = Vec::new();

    while let Some(c) = iter.next() {
        if matches!(c, '-' | '0'..='9' | 'A'..='Z' | 'a'..='z' | '_' | '/' | '.' | '\\' | '*') {
            decoded.push(c as u8)
        } else if c == '%' {
            decoded.push(
                (decode_hex(iter.next().ok_or_else(|| {
                    Error::Address("incomplete percent-encoded sequence".to_owned())
                })?)?
                    << 4)
                    | decode_hex(iter.next().ok_or_else(|| {
                        Error::Address("incomplete percent-encoded sequence".to_owned())
                    })?)?,
            );
        } else {
            return Err(Error::Address("Invalid character in address".to_owned()));
        }
    }

    Ok(decoded)
}

pub(super) fn encode_percents(f: &mut Formatter<'_>, mut value: &[u8]) -> std::fmt::Result {
    const LOOKUP: &str = "\
%00%01%02%03%04%05%06%07%08%09%0a%0b%0c%0d%0e%0f\
%10%11%12%13%14%15%16%17%18%19%1a%1b%1c%1d%1e%1f\
%20%21%22%23%24%25%26%27%28%29%2a%2b%2c%2d%2e%2f\
%30%31%32%33%34%35%36%37%38%39%3a%3b%3c%3d%3e%3f\
%40%41%42%43%44%45%46%47%48%49%4a%4b%4c%4d%4e%4f\
%50%51%52%53%54%55%56%57%58%59%5a%5b%5c%5d%5e%5f\
%60%61%62%63%64%65%66%67%68%69%6a%6b%6c%6d%6e%6f\
%70%71%72%73%74%75%76%77%78%79%7a%7b%7c%7d%7e%7f\
%80%81%82%83%84%85%86%87%88%89%8a%8b%8c%8d%8e%8f\
%90%91%92%93%94%95%96%97%98%99%9a%9b%9c%9d%9e%9f\
%a0%a1%a2%a3%a4%a5%a6%a7%a8%a9%aa%ab%ac%ad%ae%af\
%b0%b1%b2%b3%b4%b5%b6%b7%b8%b9%ba%bb%bc%bd%be%bf\
%c0%c1%c2%c3%c4%c5%c6%c7%c8%c9%ca%cb%cc%cd%ce%cf\
%d0%d1%d2%d3%d4%d5%d6%d7%d8%d9%da%db%dc%dd%de%df\
%e0%e1%e2%e3%e4%e5%e6%e7%e8%e9%ea%eb%ec%ed%ee%ef\
%f0%f1%f2%f3%f4%f5%f6%f7%f8%f9%fa%fb%fc%fd%fe%ff";

    loop {
        let pos = value.iter().position(
            |c| !matches!(c, b'-' | b'0'..=b'9' | b'A'..=b'Z' | b'a'..=b'z' | b'_' | b'/' | b'.' | b'\\' | b'*'),
        );

        if let Some(pos) = pos {
            // SAFETY: The above `position()` call made sure that only ASCII chars are in the string
            // up to `pos`
            f.write_str(unsafe { from_utf8_unchecked(&value[..pos]) })?;

            let c = value[pos];
            value = &value[pos + 1..];

            let pos = c as usize * 3;
            f.write_str(&LOOKUP[pos..pos + 3])?;
        } else {
            // SAFETY: The above `position()` call made sure that only ASCII chars are in the rest
            // of the string
            f.write_str(unsafe { from_utf8_unchecked(value) })?;
            return Ok(());
        }
    }
}

impl Display for Transport {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Tcp(tcp) => write!(f, "{tcp}")?,
            Self::Unix(unix) => write!(f, "{unix}")?,
            #[cfg(unix)]
            Self::Unixexec(unixexec) => write!(f, "{unixexec}")?,
            #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
            Self::Vsock(vsock) => write!(f, "{}", vsock)?,
            #[cfg(windows)]
            Self::Autolaunch(autolaunch) => write!(f, "{autolaunch}")?,
            #[cfg(target_os = "macos")]
            Self::Launchd(launchd) => write!(f, "{launchd}")?,
            #[cfg(unix)]
            Self::Ibus(ibus) => write!(f, "{ibus}")?,
        }

        Ok(())
    }
}