zwire_host/
proto.rs

1//! Wire transport + small reply helpers, shared by every transport.
2//!
3//! The host speaks two framings over one JSON message model:
4//!   * **Native messaging** (Chrome): a little-endian `u32` byte length followed
5//!     by a UTF-8 JSON body.
6//!   * **NDJSON** (everything else — sockets, tmux, emacs, scripts, other
7//!     languages): one JSON object per line.
8//!
9//! A [`Peer`] is the write half of a connection; it frames outgoing messages
10//! per its mode. Capabilities (RPC replies, sysinfo frames, PTY output) only
11//! ever see [`Out`] and never care which framing is in use.
12use base64::Engine;
13use serde_json::Value;
14use std::io::{self, BufRead, Read, Write};
15use std::sync::{Arc, Mutex};
16
17/// Largest inbound message we will allocate for (128 MiB). Guards against a
18/// bogus length prefix asking us to buffer the whole address space.
19const MAX_MSG: usize = 128 * 1024 * 1024;
20
21/// Outbound framing for a connection.
22#[derive(Clone, Copy)]
23pub enum Framing {
24    /// Chrome native messaging: `u32` little-endian length prefix + JSON body.
25    Native,
26    /// One compact JSON object per line.
27    Ndjson,
28}
29
30/// The write half of a connection. Cloneable via [`Out`] so background threads
31/// (sysinfo streamer, PTY reader) can push frames concurrently; the mutex keeps
32/// their frames from interleaving mid-message.
33pub struct Peer {
34    w: Mutex<Box<dyn Write + Send>>,
35    framing: Framing,
36}
37
38/// Shared handle to a [`Peer`]. Every capability writes through this.
39pub type Out = Arc<Peer>;
40
41impl Peer {
42    /// Wrap a writer with the given framing.
43    pub fn new(w: Box<dyn Write + Send>, framing: Framing) -> Out {
44        Arc::new(Peer {
45            w: Mutex::new(w),
46            framing,
47        })
48    }
49    /// Native-messaging (Chrome) sink over `stdout`.
50    pub fn native(w: Box<dyn Write + Send>) -> Out {
51        Self::new(w, Framing::Native)
52    }
53    /// NDJSON sink (sockets and CLI).
54    pub fn ndjson(w: Box<dyn Write + Send>) -> Out {
55        Self::new(w, Framing::Ndjson)
56    }
57    /// Frame and write one message. Errors when the peer has hung up, which
58    /// streaming callers use as the signal to stop.
59    pub fn send(&self, v: &Value) -> io::Result<()> {
60        let data = serde_json::to_vec(v).unwrap_or_default();
61        let mut o = self.w.lock().unwrap();
62        match self.framing {
63            Framing::Native => {
64                o.write_all(&(data.len() as u32).to_le_bytes())?;
65                o.write_all(&data)?;
66            }
67            Framing::Ndjson => {
68                o.write_all(&data)?;
69                o.write_all(b"\n")?;
70            }
71        }
72        o.flush()
73    }
74}
75
76/// Send a message on `out` (thin wrapper over [`Peer::send`]).
77pub fn send_msg(out: &Out, v: &Value) -> io::Result<()> {
78    out.send(v)
79}
80
81/// Send `v` as a reply to `req`, copying `req`'s correlation `id` (when present)
82/// onto it so multiplexed callers can match replies to requests.
83pub fn respond(out: &Out, req: &Value, mut v: Value) {
84    if let (Some(obj), Some(id)) = (v.as_object_mut(), req.get("id")) {
85        if !id.is_null() {
86            obj.insert("id".into(), id.clone());
87        }
88    }
89    // Record the response (rx) to the shared host log. Only real command replies
90    // pass through here; streaming frames (sysinfo/pty/job) use out.send directly
91    // and are intentionally not logged.
92    crate::hostlog::record("rx", req, &v);
93    let _ = out.send(&v);
94}
95
96/// Read one native-messaging (`u32`-prefixed) message. `None` on EOF, a short
97/// read, an out-of-range length, or invalid JSON — any of which ends the stream.
98pub fn read_native<R: Read>(r: &mut R) -> Option<Value> {
99    let mut len = [0u8; 4];
100    r.read_exact(&mut len).ok()?;
101    let n = u32::from_le_bytes(len) as usize;
102    if n == 0 || n > MAX_MSG {
103        return None;
104    }
105    let mut buf = vec![0u8; n];
106    r.read_exact(&mut buf).ok()?;
107    serde_json::from_slice(&buf).ok()
108}
109
110/// Read one NDJSON message: the next non-blank line parsed as JSON. `None` on
111/// EOF. Blank lines are skipped; malformed lines yield `Some(Value::Null)` so
112/// the caller can reply with an error rather than dropping the connection.
113pub fn read_ndjson<R: BufRead>(r: &mut R) -> Option<Value> {
114    loop {
115        let mut line = String::new();
116        if r.read_line(&mut line).ok()? == 0 {
117            return None;
118        }
119        let t = line.trim();
120        if t.is_empty() {
121            continue;
122        }
123        return Some(serde_json::from_str(t).unwrap_or(Value::Null));
124    }
125}
126
127/// Standard base64 encode (any binary payload crossing the JSON pipe).
128pub fn b64_encode(bytes: &[u8]) -> String {
129    base64::engine::general_purpose::STANDARD.encode(bytes)
130}
131
132/// Standard base64 decode; `None` on malformed input.
133pub fn b64_decode(s: &str) -> Option<Vec<u8>> {
134    base64::engine::general_purpose::STANDARD.decode(s).ok()
135}
zwire_host/proto.rs

zwire_host/
proto.rs
