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

teams_cli/desktop/
helper.rs
