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            path
309        }
310
311        #[tokio::test]
312        async fn handshakes_and_refuses_other_protocols_only() {
313            let dir = tempfile::tempdir().unwrap();
314            let mut running = start(&fake_helper(dir.path(), VERSION, PROTOCOL, 9))
315                .await
316                .unwrap();
317            assert_eq!(running.call(json!({"op": "x"})).await.ok(), Some(json!(1)));
318            assert_eq!(running.call(json!({"op": "x"})).await.ok(), Some(json!(2)));
319            assert!(matches!(
320                running.call(json!({"op": "fail"})).await,
321                Err(Failure::Refused(m)) if m == "refused"
322            ));
323            // Another version or build of the same protocol is accepted.
324            let older = tempfile::tempdir().unwrap();
325            assert!(
326                start(&fake_helper(older.path(), "0.0.1", PROTOCOL, 9))
327                    .await
328                    .is_ok()
329            );
330            let other = tempfile::tempdir().unwrap();
331            let error = start(&fake_helper(other.path(), "0.0.1", 1, 9))
332                .await
333                .err()
334                .unwrap()
335                .to_string();
336            assert!(
337                error
338                    .contains("is teams-cli 0.0.1 (0123abcd) speaking protocol 1, but this CLI is"),
339                "{error}"
340            );
341            assert!(
342                error.contains(&format!("speaking protocol {PROTOCOL};")),
343                "{error}"
344            );
345        }
346
347        #[tokio::test]
348        async fn a_dead_helper_is_replaced_once() {
349            let dir = tempfile::tempdir().unwrap();
350            let mut client = HelperClient {
351                running: None,
352                executable: Some(fake_helper(dir.path(), VERSION, PROTOCOL, 1)),
353            };
354            // Each helper answers once, then exits; the next call starts a new one.
355            for _ in 0..3 {
356                assert_eq!(client.call(json!({"op": "x"})).await.unwrap(), json!(1));
357            }
358            assert!(matches!(
359                client.call(json!({"op": "fail"})).await,
360                Err(AppError::Desktop(m)) if m == "refused"
361            ));
362        }
363
364        #[tokio::test]
365        async fn a_lost_answer_to_a_single_call_is_unconfirmed() {
366            let dir = tempfile::tempdir().unwrap();
367            // The helper exits right after its first answer, so it is gone for the second.
368            let mut client = HelperClient {
369                running: None,
370                executable: Some(fake_helper(dir.path(), VERSION, PROTOCOL, 1)),
371            };
372            assert_eq!(
373                client.call_once(json!({"op": "x"})).await.unwrap(),
374                json!(1)
375            );
376            assert_eq!(
377                client.call_once(json!({"op": "x"})).await.unwrap(),
378                json!({"error": "unconfirmed"})
379            );
380            // The next call starts a fresh helper rather than reusing the dead one.
381            assert_eq!(
382                client.call_once(json!({"op": "x"})).await.unwrap(),
383                json!(1)
384            );
385        }
386    }
387}
388
389#[cfg(test)]
390mod tests {
391    use super::*;
392    use crate::desktop::{owner::tests::FakeSystem, session::Session};
393    use tokio::io::{BufReader, duplex};
394
395    #[tokio::test]
396    async fn serves_hello_probe_errors_and_stops_at_end_of_input() {
397        let (client, server) = duplex(1 << 20);
398        let (server_read, server_write) = tokio::io::split(server);
399        let task = tokio::spawn(async move {
400            let mut session = Session::new(FakeSystem {
401                listeners: vec![],
402                ..FakeSystem::teams()
403            });
404            serve(BufReader::new(server_read), server_write, &mut session).await
405        });
406        let (client_read, mut client_write) = tokio::io::split(client);
407        let mut client_read = BufReader::new(client_read);
408        let mut call = async |request: Value| {
409            write_line(&mut client_write, &request).await.unwrap();
410            serde_json::from_str::<Value>(&read_line(&mut client_read).await.unwrap().unwrap())
411                .unwrap()
412        };
413        let hello = call(json!({"id": 1, "op": "hello"})).await;
414        assert_eq!(hello["id"], 1);
415        assert_eq!(
416            hello["ok"],
417            json!({"version": VERSION, "protocol": PROTOCOL, "build": BUILD})
418        );
419        let probe = call(json!({"id": 2, "op": "probe", "port": 9222})).await;
420        assert_eq!(probe["ok"]["status"], "no_listener");
421        let bad = call(json!({"id": 3, "op": "read", "request": {"port": 9222}})).await;
422        assert!(bad["error"].is_string() && bad.get("ok").is_none());
423        let send = call(json!({"id": 4, "op": "send", "request": {"port": 9222}})).await;
424        assert!(send["error"].is_string() && send.get("ok").is_none());
425        let check = call(json!({"id": 5, "op": "send_check", "port": 9222})).await;
426        assert_eq!(check["ok"]["error"], "no_listener");
427        let unknown = call(json!({"id": 6, "op": "launch"})).await;
428        assert!(unknown["error"].is_string());
429        let invalid_account =
430            call(json!({"id": 7, "op": "send_check", "port": 9222, "account_type": "all"})).await;
431        assert!(
432            invalid_account["error"]
433                .as_str()
434                .unwrap()
435                .contains("account type")
436        );
437        // Both halves share the stream; the server sees end of input once both are gone.
438        drop((client_read, client_write));
439        task.await.unwrap().unwrap();
440    }
441
442    #[tokio::test]
443    async fn oversized_lines_are_refused() {
444        let data = vec![b'x'; MAX_LINE + 1];
445        let mut reader = BufReader::new(&data[..]);
446        assert!(read_line(&mut reader).await.is_err());
447        let mut reader = BufReader::new(&b"a\nb"[..]);
448        assert_eq!(read_line(&mut reader).await.unwrap().as_deref(), Some("a"));
449        assert_eq!(read_line(&mut reader).await.unwrap().as_deref(), Some("b"));
450        assert_eq!(read_line(&mut reader).await.unwrap(), None);
451    }
452}
teams_cli/desktop/helper.rs

teams_cli/desktop/
helper.rs
