zwire_host/
transport.rs

1//! The three ways to reach the dispatcher.
2//!
3//!   * [`stdio`]  — Chrome native messaging on `stdin`/`stdout` (the default when
4//!     the browser launches the binary).
5//!   * [`serve`]  — a local-socket daemon speaking NDJSON, so tmux, emacs,
6//!     desktop apps, plugins, and any language can connect and use every
7//!     capability. Each connection gets its own [`Session`]. Backed by a Unix
8//!     domain socket on macOS/Linux and a named pipe on Windows.
9//!   * [`call`]   — a one-line client: connect, send one request, print the
10//!     reply. Lets a shell script or editor talk to the daemon trivially.
11use crate::proto::{read_native, Out, Peer};
12use crate::session::Session;
13use std::io::{self, BufRead, Write};
14use std::path::{Path, PathBuf};
15
16/// Everything the daemon needs to start: its local socket plus optional TCP
17/// peering. Built from CLI flags in [`crate::run`].
18pub struct ServeConfig {
19    /// Local endpoint (Unix socket path / Windows pipe name).
20    pub socket: PathBuf,
21    /// Optional `host:port` to listen on for peers and remote clients.
22    pub tcp: Option<String>,
23    /// Shared token required of inbound TCP connections.
24    pub token: Option<String>,
25    /// This host's advertised peer name (defaults to the hostname).
26    pub name: Option<String>,
27    /// Peer addresses to dial and keep linked for bus federation.
28    pub peers: Vec<String>,
29}
30
31/// Chrome native-messaging loop: read `u32`-framed requests from `stdin`, drive
32/// one [`Session`], until EOF (browser closed the port).
33pub fn stdio() {
34    let out = Peer::native(Box::new(io::stdout()));
35    let mut sess = Session::new();
36    let mut stdin = io::stdin();
37    while let Some(msg) = read_native(&mut stdin) {
38        sess.handle(&out, &msg);
39    }
40}
41
42/// Emit a fatal error to `stderr` and exit non-zero.
43fn die(msg: &str) -> ! {
44    let _ = writeln!(io::stderr(), "zwire-host: {msg}");
45    std::process::exit(1);
46}
47
48/// Drive one accepted NDJSON connection to completion with the given session.
49/// Shared by the local socket, the Windows pipe, and TCP peer links.
50pub(crate) fn serve_conn(mut reader: impl BufRead, out: Out, mut sess: Session) {
51    while let Some(msg) = crate::proto::read_ndjson(&mut reader) {
52        sess.handle(&out, &msg);
53    }
54    // Connection closed: `sess` drops here, tearing down its PTYs/streams/subs.
55}
56
57/// Client half: write the request, then relay reply frames to `stdout`. One
58/// frame by default; every frame until EOF when `follow` is set (streams).
59#[cfg(any(unix, windows))]
60fn run_client(reader: impl BufRead, mut writer: impl Write, request: &str, follow: bool) -> ! {
61    let line = format!("{}\n", request.trim());
62    if let Err(e) = writer
63        .write_all(line.as_bytes())
64        .and_then(|()| writer.flush())
65    {
66        die(&format!("send: {e}"));
67    }
68    let mut reader = reader;
69    let mut stdout = io::stdout();
70    loop {
71        let mut line = String::new();
72        match reader.read_line(&mut line) {
73            Ok(0) | Err(_) => break,
74            Ok(_) => {
75                if stdout
76                    .write_all(line.as_bytes())
77                    .and_then(|()| stdout.flush())
78                    .is_err()
79                {
80                    break;
81                }
82                if !follow {
83                    break; // one reply is all an RPC produces
84                }
85            }
86        }
87    }
88    std::process::exit(0);
89}
90
91/* ---- Unix domain socket (macOS / Linux) ---- */
92#[cfg(unix)]
93mod platform {
94    use super::*;
95    use std::os::unix::fs::PermissionsExt;
96    use std::os::unix::net::{UnixListener, UnixStream};
97
98    /// Run the NDJSON daemon on `sock`, one thread per connection, forever.
99    pub fn serve(sock: &Path) -> ! {
100        if let Some(parent) = sock.parent() {
101            let _ = std::fs::create_dir_all(parent);
102            let _ = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700));
103        }
104        // A stale socket file from a previous run blocks bind(); clear it.
105        let _ = std::fs::remove_file(sock);
106        let listener = match UnixListener::bind(sock) {
107            Ok(l) => l,
108            Err(e) => die(&format!("bind {}: {e}", sock.display())),
109        };
110        // Owner-only: the socket exposes exec/fs/pty, so no other local user may
111        // connect.
112        let _ = std::fs::set_permissions(sock, std::fs::Permissions::from_mode(0o600));
113        let _ = writeln!(io::stderr(), "zwire-host: listening on {}", sock.display());
114
115        for conn in listener.incoming() {
116            match conn {
117                Ok(stream) => {
118                    std::thread::spawn(move || handle(stream));
119                }
120                Err(e) => {
121                    let _ = writeln!(io::stderr(), "zwire-host: accept: {e}");
122                }
123            }
124        }
125        std::process::exit(0);
126    }
127
128    fn handle(stream: UnixStream) {
129        let Ok(rclone) = stream.try_clone() else {
130            return;
131        };
132        let out = Peer::ndjson(Box::new(stream));
133        // Local socket clients are trusted (owner-only), so no auth is required.
134        serve_conn(io::BufReader::new(rclone), out, Session::new());
135    }
136
137    /// Connect to the daemon and run one request/reply exchange.
138    pub fn call(sock: &Path, request: &str, follow: bool) -> ! {
139        let stream = match UnixStream::connect(sock) {
140            Ok(s) => s,
141            Err(e) => die(&format!(
142                "connect {}: {e} (is the daemon running? `zwire-host serve`)",
143                sock.display()
144            )),
145        };
146        let rclone = match stream.try_clone() {
147            Ok(r) => r,
148            Err(e) => die(&format!("clone: {e}")),
149        };
150        run_client(io::BufReader::new(rclone), stream, request, follow);
151    }
152}
153
154/* ---- named pipe (Windows) ---- */
155#[cfg(windows)]
156mod platform {
157    use super::*;
158    use interprocess::local_socket::{prelude::*, GenericNamespaced, ListenerOptions, Stream};
159
160    /// Derive the pipe's namespaced name from the endpoint's leaf token. On
161    /// Windows this maps to `\\.\pipe\<name>`.
162    fn pipe_name(sock: &Path) -> String {
163        sock.file_name()
164            .map(|s| s.to_string_lossy().into_owned())
165            .filter(|s| !s.is_empty())
166            .unwrap_or_else(|| "zwire-host".to_string())
167    }
168
169    pub fn serve(sock: &Path) -> ! {
170        let raw = pipe_name(sock);
171        let name = match raw.clone().to_ns_name::<GenericNamespaced>() {
172            Ok(n) => n,
173            Err(e) => die(&format!("bad pipe name {raw}: {e}")),
174        };
175        let listener = match ListenerOptions::new().name(name).create_sync() {
176            Ok(l) => l,
177            Err(e) => die(&format!(r"bind \\.\pipe\{raw}: {e}")),
178        };
179        let _ = writeln!(io::stderr(), r"zwire-host: listening on \\.\pipe\{raw}");
180
181        loop {
182            match listener.accept() {
183                Ok(stream) => {
184                    std::thread::spawn(move || handle(stream));
185                }
186                Err(e) => {
187                    let _ = writeln!(io::stderr(), "zwire-host: accept: {e}");
188                }
189            }
190        }
191    }
192
193    fn handle(stream: Stream) {
194        let (recv, send) = stream.split();
195        let out = Peer::ndjson(Box::new(send));
196        serve_conn(io::BufReader::new(recv), out, Session::new());
197    }
198
199    pub fn call(sock: &Path, request: &str, follow: bool) -> ! {
200        let raw = pipe_name(sock);
201        let name = match raw.clone().to_ns_name::<GenericNamespaced>() {
202            Ok(n) => n,
203            Err(e) => die(&format!("bad pipe name {raw}: {e}")),
204        };
205        let stream = match Stream::connect(name) {
206            Ok(s) => s,
207            Err(e) => die(&format!(
208                r"connect \\.\pipe\{raw}: {e} (is the daemon running? `zwire-host serve`)"
209            )),
210        };
211        let (recv, send) = stream.split();
212        run_client(io::BufReader::new(recv), send, request, follow);
213    }
214}
215
216/* ---- platforms with neither Unix sockets nor named pipes ---- */
217#[cfg(not(any(unix, windows)))]
218mod platform {
219    use super::*;
220    pub fn serve(_sock: &Path) -> ! {
221        die("`serve` is not supported on this platform");
222    }
223    pub fn call(_sock: &Path, _request: &str, _follow: bool) -> ! {
224        die("`call` is not supported on this platform");
225    }
226}
227
228pub use platform::call;
229
230/// Run the daemon: configure peering, start the optional TCP listener and any
231/// outbound peer links, then run the local socket/pipe accept loop forever.
232pub fn serve(cfg: ServeConfig) -> ! {
233    crate::peer::configure(cfg.token, cfg.name);
234    if let Some(addr) = cfg.tcp {
235        crate::peer::listen_tcp(addr);
236    }
237    for peer in cfg.peers {
238        crate::peer::dial(peer);
239    }
240    platform::serve(&cfg.socket)
241}
zwire_host/transport.rs

zwire_host/
transport.rs
