teams_cli/desktop/
helper.rs

1//! The WSL bridge: the Linux CLI keeps one `teams.exe desktop-helper` running and talks
2//! newline-delimited JSON over its stdin and stdout. The helper exits when stdin closes.
3//! The protocol number is the whole contract: a helper is accepted when it speaks this
4//! CLI's protocol, whatever its version or build, which are only reported. Bump it when
5//! either side would misread the other (`desktop::tests::the_protocol_is_pinned`).
6//! `hello` and its `version` and `protocol` fields never change. Protocol 2 adds sending;
7//! protocol 4 adds recipient search and once-only chat creation.
8//! protocol 3 adds explicit account-type selection to reads, sends and send checks.
9use crate::error::AppError;
10use serde_json::{Value, json};
11use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt};
12
13pub const PROTOCOL: u32 = 4;
14const VERSION: &str = env!("CARGO_PKG_VERSION");
15/// The commit this binary was built from (see `build.rs`), reported in `hello` for
16/// diagnosis and never compared; empty when unknown.
17const BUILD: &str = env!("TEAMS_BUILD");
18const MAX_LINE: usize = super::MAX_READ + 64 * 1024;
19
20/// Reads one line, refusing lines beyond `MAX_LINE`; `None` at end of input.
21async fn read_line(reader: &mut (impl AsyncBufRead + Unpin)) -> Result<Option<String>, AppError> {
22    let mut line = Vec::new();
23    loop {
24        let buffer = reader.fill_buf().await?;
25        if buffer.is_empty() {
26            return Ok((!line.is_empty()).then(|| String::from_utf8_lossy(&line).into_owned()));
27        }
28        let (chunk, done) = match buffer.iter().position(|&b| b == b'\n') {
29            Some(end) => (&buffer[..end], Some(end + 1)),
30            None => (buffer, None),
31        };
32        if line.len() + chunk.len() > MAX_LINE {
33            return Err(AppError::Desktop(
34                "desktop helper message is too large".into(),
35            ));
36        }
37        line.extend_from_slice(chunk);
38        let consumed = done.unwrap_or(chunk.len());
39        reader.consume(consumed);
40        if done.is_some() {
41            return Ok(Some(String::from_utf8_lossy(&line).into_owned()));
42        }
43    }
44}
45
46async fn write_line(writer: &mut (impl AsyncWrite + Unpin), value: &Value) -> Result<(), AppError> {
47    let mut line = value.to_string();
48    line.push('\n');
49    writer.write_all(line.as_bytes()).await?;
50    writer.flush().await?;
51    Ok(())
52}
53
54/// The helper side: answers requests in order until the input closes.
55#[cfg(any(windows, test))]
56pub async fn serve<S: super::session::Verifier>(
57    mut reader: impl AsyncBufRead + Unpin,
58    mut writer: impl AsyncWrite + Unpin,
59    session: &mut super::session::Session<S>,
60) -> Result<(), AppError> {
61    while let Some(line) = read_line(&mut reader).await? {
62        let request: Value = serde_json::from_str(&line).unwrap_or_default();
63        let result = match request["op"].as_str() {
64            Some("hello") => Ok(json!({"version": VERSION, "protocol": PROTOCOL, "build": BUILD})),
65            Some("probe") => match request["port"].as_u64().and_then(|p| u16::try_from(p).ok()) {
66                Some(port) => session.probe(port).await.map(|metadata| json!(metadata)),
67                None => Err(AppError::InvalidInput("invalid desktop port".into())),
68            },
69            Some("read") => session.read(&request["request"]).await,
70            Some("people") => session.people(&request["request"]).await,
71            Some("send") => session.send(&request["request"]).await,
72            Some("send_check") => {
73                match request["port"].as_u64().and_then(|p| u16::try_from(p).ok()) {
74                    Some(port) => {
75                        match crate::config::DesktopAccount::from_request(&request["account_type"])
76                        {
77                            Ok(account) => session.send_check(port, account).await,
78                            Err(error) => Err(error),
79                        }
80                    }
81                    None => Err(AppError::InvalidInput("invalid desktop port".into())),
82                }
83            }
84            _ => Err(AppError::InvalidInput(
85                "unknown desktop helper request".into(),
86            )),
87        };
88        let reply = match result {
89            Ok(value) => json!({"id": request["id"], "ok": value}),
90            Err(error) => json!({"id": request["id"], "error": error.to_string()}),
91        };
92        write_line(&mut writer, &reply).await?;
93    }
94    Ok(())
95}
96
97#[cfg(not(windows))]
98pub use client::HelperClient;
99
100#[cfg(not(windows))]
101mod client {
102    use super::*;
103    use std::{
104        path::{Path, PathBuf},
105        process::Stdio,
106        time::Duration,
107    };
108    use tokio::{
109        io::BufReader,
110        process::{Child, ChildStdin, ChildStdout, Command},
111    };
112
113    const START_TIMEOUT: Duration = Duration::from_secs(20);
114    const CALL_TIMEOUT: Duration = Duration::from_secs(45);
115
116    enum Failure {
117        /// The helper answered with an error; a new helper would answer the same.
118        Refused(String),
119        /// The helper is gone or unusable.
120        Transport,
121    }
122
123    struct Running {
124        // Held so the helper is killed when this connection is dropped.
125        _child: Child,
126        stdin: ChildStdin,
127        stdout: BufReader<ChildStdout>,
128        next_id: u64,
129    }
130
131    impl Running {
132        async fn call(&mut self, mut request: Value) -> Result<Value, Failure> {
133            self.next_id += 1;
134            request["id"] = json!(self.next_id);
135            let exchange = async {
136                write_line(&mut self.stdin, &request).await.ok()?;
137                let line = read_line(&mut self.stdout).await.ok()??;
138                serde_json::from_str::<Value>(&line).ok()
139            };
140            let reply = tokio::time::timeout(CALL_TIMEOUT, exchange)
141                .await
142                .ok()
143                .flatten()
144                .ok_or(Failure::Transport)?;
145            if reply["id"] != self.next_id {
146                return Err(Failure::Transport);
147            }
148            match (reply.get("ok"), reply["error"].as_str()) {
149                (Some(ok), _) => Ok(ok.clone()),
150                (None, Some(message)) => Err(Failure::Refused(message.into())),
151                _ => Err(Failure::Transport),
152            }
153        }
154    }
155
156    /// One helper per CLI process, started on first use and restarted once if it dies.
157    #[derive(Default)]
158    pub struct HelperClient {
159        running: Option<Running>,
160        /// Resolved, and installed if needed, on first start.
161        executable: Option<PathBuf>,
162    }
163
164    impl HelperClient {
165        pub async fn call(&mut self, request: Value) -> Result<Value, AppError> {
166            if let Some(running) = &mut self.running {
167                match running.call(request.clone()).await {
168                    Ok(value) => return Ok(value),
169                    Err(Failure::Refused(message)) => return Err(AppError::Desktop(message)),
170                    Err(Failure::Transport) => self.running = None,
171                }
172            }
173            match self.started().await?.call(request).await {
174                Ok(value) => Ok(value),
175                Err(Failure::Refused(message)) => Err(AppError::Desktop(message)),
176                Err(Failure::Transport) => {
177                    self.running = None;
178                    Err(AppError::Desktop(
179                        "the Windows desktop helper stopped unexpectedly".into(),
180                    ))
181                }
182            }
183        }
184
185        /// Sends `request` exactly once. A lost answer becomes `{"error": "unconfirmed"}`
186        /// instead of a retry, because the helper may already have acted on it.
187        pub async fn call_once(&mut self, request: Value) -> Result<Value, AppError> {
188            match self.started().await?.call(request).await {
189                Ok(value) => Ok(value),
190                Err(Failure::Refused(message)) => Err(AppError::Desktop(message)),
191                Err(Failure::Transport) => {
192                    self.running = None;
193                    Ok(json!({"error": "unconfirmed"}))
194                }
195            }
196        }
197
198        /// The running helper, started first if needed.
199        async fn started(&mut self) -> Result<&mut Running, AppError> {
200            let running = match self.running.take() {
201                Some(running) => running,
202                None => {
203                    let executable = match &self.executable {
204                        Some(executable) => executable.clone(),
205                        None => super::super::install::ensure().await?,
206                    };
207                    let running = start(&executable).await?;
208                    self.executable = Some(executable);
209                    running
210                }
211            };
212            Ok(self.running.insert(running))
213        }
214    }
215
216    /// Why a helper that answered `hello` cannot serve this CLI, if it cannot.
217    fn mismatch(hello: &Value) -> Option<String> {
218        (hello["protocol"] != PROTOCOL).then(|| {
219            let theirs = describe(
220                hello["version"].as_str(),
221                hello["build"].as_str(),
222                hello["protocol"].as_u64(),
223            );
224            let ours = describe(Some(VERSION), Some(BUILD), Some(PROTOCOL.into()));
225            format!("is {theirs}, but this CLI is {ours}")
226        })
227    }
228
229    /// `teams-cli 0.1.9 (abc123) speaking protocol 2`, leaving out what is unknown.
230    fn describe(version: Option<&str>, build: Option<&str>, protocol: Option<u64>) -> String {
231        let mut text = format!("teams-cli {}", version.unwrap_or("of an unknown version"));
232        if let Some(build) = build.filter(|b| !b.is_empty()) {
233            text.push_str(&format!(" ({build})"));
234        }
235        match protocol {
236            Some(protocol) => text.push_str(&format!(" speaking protocol {protocol}")),
237            None => text.push_str(" without a protocol number"),
238        }
239        text
240    }
241
242    async fn start(executable: &Path) -> Result<Running, AppError> {
243        let mut child = Command::new(executable)
244            .arg("desktop-helper")
245            .stdin(Stdio::piped())
246            .stdout(Stdio::piped())
247            .stderr(Stdio::null())
248            .kill_on_drop(true)
249            .spawn()
250            .map_err(|_| {
251                AppError::Desktop(format!(
252                    "could not start the Windows desktop helper at {}; check that WSL can run Windows executables",
253                    executable.display()
254                ))
255            })?;
256        let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else {
257            return Err(AppError::Desktop(
258                "desktop helper pipes are unavailable".into(),
259            ));
260        };
261        let mut running = Running {
262            _child: child,
263            stdin,
264            stdout: BufReader::new(stdout),
265            next_id: 0,
266        };
267        let hello = tokio::time::timeout(START_TIMEOUT, running.call(json!({"op": "hello"})))
268            .await
269            .ok()
270            .and_then(Result::ok);
271        let problem = match &hello {
272            None => Some("did not answer".to_owned()),
273            Some(hello) => mismatch(hello),
274        };
275        if let Some(problem) = problem {
276            return Err(AppError::Desktop(format!(
277                "the Windows desktop helper at {} {problem}; {}",
278                executable.display(),
279                super::super::install::remedy()
280            )));
281        }
282        Ok(running)
283    }
284
285    #[cfg(all(test, unix))]
286    mod tests {
287        use super::*;
288        use std::os::unix::fs::PermissionsExt;
289
290        /// A stand-in helper: answers hello with `version`, then echoes requests, and
291        /// exits after `lives` echoes.
292        fn fake_helper(dir: &std::path::Path, version: &str, protocol: u32, lives: u32) -> PathBuf {
293            let path = dir.join("teams.exe");
294            let script = format!(
295                r#"#!/bin/sh
296[ "$1" = desktop-helper ] || exit 2
297n=0
298while read -r line; do
299  id=$(printf '%s' "$line" | sed 's/.*"id":\([0-9]*\).*/\1/')
300  case "$line" in
301    *'"op":"hello"'*) echo "{{\"id\":$id,\"ok\":{{\"version\":\"{version}\",\"protocol\":{protocol},\"build\":\"0123abcd\"}}}}" ;;
302    *'"op":"fail"'*) echo "{{\"id\":$id,\"error\":\"refused\"}}" ;;
303    *) n=$((n+1)); echo "{{\"id\":$id,\"ok\":$n}}"; [ $n -ge {lives} ] && exit 0 ;;
304  esac
305done
306"#
307            );
308            std::fs::write(&path, script).unwrap();
309            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
310            // A process forked by another test while the script was open for writing holds
311            // that handle until it execs, and executing the script meanwhile fails with
312            // "text file busy". Wait that out here, so the test's own start never sees it.
313            while matches!(
314                std::process::Command::new(&path).status(),
315                Err(e) if e.kind() == std::io::ErrorKind::ExecutableFileBusy
316            ) {
317                std::thread::sleep(Duration::from_millis(5));
318            }
319            path
320        }
321
322        #[tokio::test]
323        async fn handshakes_and_refuses_other_protocols_only() {
324            let dir = tempfile::tempdir().unwrap();
325            let mut running = start(&fake_helper(dir.path(), VERSION, PROTOCOL, 9))
326                .await
327                .unwrap();
328            assert_eq!(running.call(json!({"op": "x"})).await.ok(), Some(json!(1)));
329            assert_eq!(running.call(json!({"op": "x"})).await.ok(), Some(json!(2)));
330            assert!(matches!(
331                running.call(json!({"op": "fail"})).await,
332                Err(Failure::Refused(m)) if m == "refused"
333            ));
334            // Another version or build of the same protocol is accepted.
335            let older = tempfile::tempdir().unwrap();
336            assert!(
337                start(&fake_helper(older.path(), "0.0.1", PROTOCOL, 9))
338                    .await
339                    .is_ok()
340            );
341            let other = tempfile::tempdir().unwrap();
342            let error = start(&fake_helper(other.path(), "0.0.1", 1, 9))
343                .await
344                .err()
345                .unwrap()
346                .to_string();
347            assert!(
348                error
349                    .contains("is teams-cli 0.0.1 (0123abcd) speaking protocol 1, but this CLI is"),
350                "{error}"
351            );
352            assert!(
353                error.contains(&format!("speaking protocol {PROTOCOL};")),
354                "{error}"
355            );
356        }
357
358        #[tokio::test]
359        async fn a_dead_helper_is_replaced_once() {
360            let dir = tempfile::tempdir().unwrap();
361            let mut client = HelperClient {
362                running: None,
363                executable: Some(fake_helper(dir.path(), VERSION, PROTOCOL, 1)),
364            };
365            // Each helper answers once, then exits; the next call starts a new one.
366            for _ in 0..3 {
367                assert_eq!(client.call(json!({"op": "x"})).await.unwrap(), json!(1));
368            }
369            assert!(matches!(
370                client.call(json!({"op": "fail"})).await,
371                Err(AppError::Desktop(m)) if m == "refused"
372            ));
373        }
374
375        #[tokio::test]
376        async fn a_lost_answer_to_a_single_call_is_unconfirmed() {
377            let dir = tempfile::tempdir().unwrap();
378            // The helper exits right after its first answer, so it is gone for the second.
379            let mut client = HelperClient {
380                running: None,
381                executable: Some(fake_helper(dir.path(), VERSION, PROTOCOL, 1)),
382            };
383            assert_eq!(
384                client.call_once(json!({"op": "x"})).await.unwrap(),
385                json!(1)
386            );
387            assert_eq!(
388                client.call_once(json!({"op": "x"})).await.unwrap(),
389                json!({"error": "unconfirmed"})
390            );
391            // The next call starts a fresh helper rather than reusing the dead one.
392            assert_eq!(
393                client.call_once(json!({"op": "x"})).await.unwrap(),
394                json!(1)
395            );
396        }
397    }
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403    use crate::desktop::{owner::tests::FakeSystem, session::Session};
404    use tokio::io::{BufReader, duplex};
405
406    #[tokio::test]
407    async fn serves_hello_probe_errors_and_stops_at_end_of_input() {
408        let (client, server) = duplex(1 << 20);
409        let (server_read, server_write) = tokio::io::split(server);
410        let task = tokio::spawn(async move {
411            let mut session = Session::new(FakeSystem {
412                listeners: vec![],
413                ..FakeSystem::teams()
414            });
415            serve(BufReader::new(server_read), server_write, &mut session).await
416        });
417        let (client_read, mut client_write) = tokio::io::split(client);
418        let mut client_read = BufReader::new(client_read);
419        let mut call = async |request: Value| {
420            write_line(&mut client_write, &request).await.unwrap();
421            serde_json::from_str::<Value>(&read_line(&mut client_read).await.unwrap().unwrap())
422                .unwrap()
423        };
424        let hello = call(json!({"id": 1, "op": "hello"})).await;
425        assert_eq!(hello["id"], 1);
426        assert_eq!(
427            hello["ok"],
428            json!({"version": VERSION, "protocol": PROTOCOL, "build": BUILD})
429        );
430        let probe = call(json!({"id": 2, "op": "probe", "port": 9222})).await;
431        assert_eq!(probe["ok"]["status"], "no_listener");
432        let bad = call(json!({"id": 3, "op": "read", "request": {"port": 9222}})).await;
433        assert!(bad["error"].is_string() && bad.get("ok").is_none());
434        let send = call(json!({"id": 4, "op": "send", "request": {"port": 9222}})).await;
435        assert!(send["error"].is_string() && send.get("ok").is_none());
436        let check = call(json!({"id": 5, "op": "send_check", "port": 9222})).await;
437        assert_eq!(check["ok"]["error"], "no_listener");
438        let unknown = call(json!({"id": 6, "op": "launch"})).await;
439        assert!(unknown["error"].is_string());
440        let invalid_account =
441            call(json!({"id": 7, "op": "send_check", "port": 9222, "account_type": "all"})).await;
442        assert!(
443            invalid_account["error"]
444                .as_str()
445                .unwrap()
446                .contains("account type")
447        );
448        // Both halves share the stream; the server sees end of input once both are gone.
449        drop((client_read, client_write));
450        task.await.unwrap().unwrap();
451    }
452
453    #[tokio::test]
454    async fn oversized_lines_are_refused() {
455        let data = vec![b'x'; MAX_LINE + 1];
456        let mut reader = BufReader::new(&data[..]);
457        assert!(read_line(&mut reader).await.is_err());
458        let mut reader = BufReader::new(&b"a\nb"[..]);
459        assert_eq!(read_line(&mut reader).await.unwrap().as_deref(), Some("a"));
460        assert_eq!(read_line(&mut reader).await.unwrap().as_deref(), Some("b"));
461        assert_eq!(read_line(&mut reader).await.unwrap(), None);
462    }
463}
teams_cli/desktop/helper.rs

teams_cli/desktop/
helper.rs
