Skip to main content

max / makenotwork

29.2 KB · 779 lines History Blame Raw
1 //! Streaming command execution — local or over SSH.
2 //!
3 //! The single most valuable primitive both tools share. Sando's `deploy.rs`
4 //! shelled out to `ssh`/`rsync` with buffered `.output()`; an app-build
5 //! orchestrator instead needs the merged stdout+stderr streamed live (a
6 //! `cargo tauri build` writes progress to stderr for minutes), so this is a
7 //! proper streaming runner rather than a buffered one.
8 //!
9 //! A [`RemoteHost`] is either the local machine (`ssh_target == "local"` or
10 //! empty) or a tailnet host reached over SSH. [`RemoteHost::run_streaming`]
11 //! spawns the command, drains both pipes concurrently into a shared
12 //! [`LogSink`], and returns the exit status plus the full captured bytes so a
13 //! caller can still post-process the whole output (classify, grep a success
14 //! banner, etc.).
15 //!
16 //! This is the low-level transport primitive. The capability-gated
17 //! [`crate::Executor`] trait ([`crate::LocalExec`] / [`crate::SshExec`]) is the
18 //! layer built on top of it.
19
20 use anyhow::{Context, Result};
21 use async_trait::async_trait;
22 use std::process::{ExitStatus, Stdio};
23 use std::sync::Arc;
24 use tokio::io::{AsyncRead, AsyncReadExt};
25 use tokio::process::Command;
26 use tokio::sync::Mutex;
27
28 /// SSH options used everywhere we shell out to ssh — fail fast, no prompts.
29 /// Matches Sando's original `deploy.rs` so behavior is identical across tools.
30 pub const SSH_FLAGS: &[&str] = &[
31 "-o",
32 "BatchMode=yes",
33 "-o",
34 "ConnectTimeout=10",
35 "-o",
36 "StrictHostKeyChecking=accept-new",
37 ];
38
39 /// A sink for streamed command output. Each chunk reflects a tokio read
40 /// boundary — chunks are NOT line-aligned; consumers that want lines must
41 /// reassemble. `ops_core::live_log::LiveLog` is the canonical implementation
42 /// (append-to-disk + broadcast).
43 ///
44 /// `#[async_trait]` keeps the trait object-safe so it can be passed as
45 /// `&mut dyn LogSink` into the [`crate::Executor`] trait while still being
46 /// usable behind an `Arc<Mutex<_>>` by [`RemoteHost::run_streaming`].
47 #[async_trait]
48 pub trait LogSink: Send {
49 async fn write_chunk(&mut self, bytes: &[u8]);
50 }
51
52 /// A build host: the local machine or an SSH target (a tailnet alias such as
53 /// `mbp`, or `user@host`), optionally on a non-default SSH port.
54 #[derive(Debug, Clone)]
55 pub struct RemoteHost {
56 ssh_target: String,
57 /// `None` = ssh's default (22, or whatever `~/.ssh/config` says for this
58 /// target). The port lives here rather than on one transport so the exec
59 /// path (`ssh -p`) and the sync path (`rsync -e "ssh -p"`) can never
60 /// disagree about which port a host is on.
61 port: Option<u16>,
62 }
63
64 /// Result of a streamed run: the exit status plus a bounded recent tail of the
65 /// captured stdout and stderr (each already forwarded to the sink — and, for
66 /// the engine, the on-disk log — byte-exact as it arrived). The returned buffer
67 /// is only for callers that branch on output, so it is capped at
68 /// [`OUTPUT_TAIL_CAP`] to avoid holding a multi-GB build log in RAM.
69 #[derive(Debug)]
70 pub struct RunOutput {
71 pub status: ExitStatus,
72 pub stdout: Vec<u8>,
73 pub stderr: Vec<u8>,
74 }
75
76 impl RunOutput {
77 pub fn success(&self) -> bool {
78 self.status.success()
79 }
80 }
81
82 /// Cap for the in-memory capture returned in [`RunOutput`]. The live sink and
83 /// the on-disk log carry the byte-exact stream; this buffer is a recent tail.
84 pub(crate) const OUTPUT_TAIL_CAP: usize = 256 * 1024;
85
86 /// Append `chunk`, then trim the front so `buf` retains at most its last `cap`
87 /// bytes — a rolling tail that bounds memory for verbose, long-running commands.
88 pub(crate) fn push_bounded(buf: &mut Vec<u8>, chunk: &[u8], cap: usize) {
89 buf.extend_from_slice(chunk);
90 if buf.len() > cap {
91 buf.drain(..buf.len() - cap);
92 }
93 }
94
95 /// Rebuild an `ExitStatus` from a plain process exit code (0..=255).
96 ///
97 /// On Unix the code lives in the high byte of the wait-status encoding. All a
98 /// recipe branches on — `.success()` and the non-zero distinction — round-trips
99 /// exactly.
100 #[cfg(unix)]
101 pub(crate) fn exit_status_from_code(code: i32) -> ExitStatus {
102 use std::os::unix::process::ExitStatusExt;
103 ExitStatus::from_raw((code & 0xff) << 8)
104 }
105
106 #[cfg(not(unix))]
107 pub(crate) fn exit_status_from_code(code: i32) -> ExitStatus {
108 use std::os::windows::process::ExitStatusExt;
109 ExitStatus::from_raw(code as u32)
110 }
111
112 /// A per-run marker used to carry a remote command's real exit code back to us
113 /// in its own stdout.
114 ///
115 /// **Why this exists.** `ssh` is supposed to relay the remote command's exit
116 /// status, and a regular sshd does. Tailscale SSH does not: it closes the
117 /// channel without an `exit-status` message, so the local `ssh` exits 0 no
118 /// matter how the remote command ended. Verified 2026-07-16 — `ssh mbp 'exit
119 /// 42'` reports 0 while `ssh astra 'exit 42'` (regular sshd) reports 42. Neither
120 /// `-t`, `-T`, nor a `bash -c` wrapper changes it. That turns every remote
121 /// failure into a silent success, which for a release pipeline means shipping a
122 /// green step that produced no artifact.
123 ///
124 /// So we stop trusting the transport's status and have the remote shell tell us
125 /// the code in-band: [`RcSentinel::wrap`] appends a `printf` of `$?` to the
126 /// script, and [`RcFilter`] pulls that line back out of the stream before anyone
127 /// sees it. When the sentinel is missing we fail closed rather than guess.
128 ///
129 /// The marker carries a per-run nonce so build output cannot forge it, and the
130 /// real sentinel is always the last thing on stdout, so parsing is anchored to
131 /// the end of the stream rather than searching the whole log.
132 #[derive(Debug, Clone)]
133 pub(crate) struct RcSentinel {
134 /// `__ops_exec_rc_<nonce>=`, the shell-safe body of the marker.
135 body: String,
136 /// `\n__ops_exec_rc_<nonce>=` — what to scan the byte stream for.
137 marker: Vec<u8>,
138 }
139
140 impl RcSentinel {
141 pub(crate) fn new() -> Self {
142 let body = format!("__ops_exec_rc_{}=", nonce());
143 let marker = format!("\n{body}").into_bytes();
144 Self { body, marker }
145 }
146
147 /// Wrap `script` so its exit code is printed as the final line of stdout.
148 ///
149 /// The script runs in a **subshell**, which is load-bearing: our callers
150 /// send `set -e` scripts, and an `exit`/errexit at the top level of the
151 /// remote shell would take the whole shell down before the `printf` ran,
152 /// costing us the sentinel on exactly the failures we care about most. A
153 /// subshell contains both, and the parent (which never sets `-e`) lives to
154 /// report `$?`.
155 pub(crate) fn wrap(&self, script: &str) -> String {
156 format!(
157 "(\n{script}\n)\n__ops_exec_status=$?\nprintf '\\n{}%s\\n' \"$__ops_exec_status\"\n",
158 self.body
159 )
160 }
161
162 /// How many trailing bytes must be held back to be sure a complete sentinel
163 /// is never split across the boundary: the marker, the digits, and the
164 /// closing newline.
165 fn tail_len(&self) -> usize {
166 self.marker.len() + 16
167 }
168
169 /// Split a stream tail into (bytes that are real output, the parsed code).
170 /// Returns `None` for the code when no complete sentinel is present.
171 fn split<'a>(&self, tail: &'a [u8]) -> (&'a [u8], Option<i32>) {
172 let Some(idx) = last_index_of(tail, &self.marker) else {
173 return (tail, None);
174 };
175 let rest = &tail[idx + self.marker.len()..];
176 let Some(end) = rest.iter().position(|b| *b == b'\n') else {
177 return (tail, None);
178 };
179 // The sentinel is the last thing written; anything after it means this
180 // is not the line we appended.
181 if !rest[end + 1..].is_empty() {
182 return (tail, None);
183 }
184 match std::str::from_utf8(&rest[..end])
185 .ok()
186 .and_then(|s| s.trim().parse::<i32>().ok())
187 {
188 Some(code) => (&tail[..idx], Some(code)),
189 None => (tail, None),
190 }
191 }
192 }
193
194 /// A short, shell-safe, per-run token. Time plus a counter is plenty: this only
195 /// has to be unpredictable to the *script*, not to an attacker with our source.
196 fn nonce() -> String {
197 use std::sync::atomic::{AtomicU64, Ordering};
198 static COUNTER: AtomicU64 = AtomicU64::new(0);
199 let n = COUNTER.fetch_add(1, Ordering::Relaxed);
200 let t = std::time::SystemTime::now()
201 .duration_since(std::time::UNIX_EPOCH)
202 .map_or(0, |d| d.as_nanos() as u64);
203 format!("{t:016x}{:04x}", n & 0xffff)
204 }
205
206 fn last_index_of(haystack: &[u8], needle: &[u8]) -> Option<usize> {
207 if needle.is_empty() || haystack.len() < needle.len() {
208 return None;
209 }
210 (0..=haystack.len() - needle.len())
211 .rev()
212 .find(|&i| &haystack[i..i + needle.len()] == needle)
213 }
214
215 /// Strips an [`RcSentinel`] off the tail of a stdout stream while it is still
216 /// being streamed.
217 ///
218 /// Holds back the last [`RcSentinel::tail_len`] bytes so the sentinel is never
219 /// forwarded to the sink (the operator's live log) or into the captured buffer
220 /// that callers grep for banners. Only the tail is delayed, and only until the
221 /// stream ends, so live output is unaffected in practice.
222 ///
223 /// `None` = a transport that reports exit status honestly (local `sh -c`), where
224 /// there is no sentinel and every byte passes straight through.
225 pub(crate) struct RcFilter {
226 sentinel: Option<RcSentinel>,
227 hold: Vec<u8>,
228 }
229
230 impl RcFilter {
231 pub(crate) fn new(sentinel: Option<RcSentinel>) -> Self {
232 Self {
233 sentinel,
234 hold: Vec::new(),
235 }
236 }
237
238 /// Feed a freshly-read chunk; returns the bytes that are safe to forward now.
239 pub(crate) fn feed(&mut self, chunk: &[u8]) -> Vec<u8> {
240 let Some(s) = &self.sentinel else {
241 return chunk.to_vec();
242 };
243 self.hold.extend_from_slice(chunk);
244 let cap = s.tail_len();
245 if self.hold.len() > cap {
246 let cut = self.hold.len() - cap;
247 self.hold.drain(..cut).collect()
248 } else {
249 Vec::new()
250 }
251 }
252
253 /// End of stream: returns the last real output bytes plus the parsed code.
254 pub(crate) fn finish(self) -> (Vec<u8>, Option<i32>) {
255 match &self.sentinel {
256 None => (self.hold, None),
257 Some(s) => {
258 let (rest, code) = s.split(&self.hold);
259 (rest.to_vec(), code)
260 }
261 }
262 }
263 }
264
265 /// Decide the status to report from what the transport claimed and what the
266 /// remote shell actually said.
267 ///
268 /// The sentinel wins whenever we have it: it is the remote command's own `$?`,
269 /// while `status` is only the transport's opinion of it. With no sentinel, a
270 /// non-zero transport status is still a real failure worth surfacing (ssh exits
271 /// 255 when it cannot connect, and no sentinel is written). A *successful*
272 /// transport status with no sentinel is the dangerous case this whole mechanism
273 /// exists for, so it is an error rather than a green step.
274 pub(crate) fn resolve_remote_status(
275 host: &str,
276 status: ExitStatus,
277 sentinel_code: Option<i32>,
278 ) -> Result<ExitStatus> {
279 if let Some(code) = sentinel_code {
280 return Ok(exit_status_from_code(code));
281 }
282 if !status.success() {
283 return Ok(status);
284 }
285 anyhow::bail!(
286 "{host}: the remote command produced no exit-status sentinel, and ssh reported success — \
287 so its real exit code is unknown and cannot be trusted. This is what Tailscale SSH does \
288 (it closes the channel without an exit-status message, making every command look like it \
289 passed); it also happens if the remote shell is not POSIX or the connection dropped \
290 mid-stream. Check `ssh {host} 'exit 42'`: a healthy host reports 42."
291 )
292 }
293
294 impl RemoteHost {
295 /// `"local"` (or empty) runs commands directly via `sh -c`; anything else
296 /// is an SSH target, on ssh's default port until [`RemoteHost::with_port`].
297 pub fn new(ssh_target: impl Into<String>) -> Self {
298 Self {
299 ssh_target: ssh_target.into(),
300 port: None,
301 }
302 }
303
304 /// Reach this host on a non-default SSH port. `None` keeps ssh's default,
305 /// so a caller with an `Option<u16>` can pass it straight through.
306 #[must_use]
307 pub fn with_port(mut self, port: Option<u16>) -> Self {
308 self.port = port;
309 self
310 }
311
312 pub fn is_local(&self) -> bool {
313 self.ssh_target == "local" || self.ssh_target.is_empty()
314 }
315
316 pub fn ssh_target(&self) -> &str {
317 &self.ssh_target
318 }
319
320 pub fn port(&self) -> Option<u16> {
321 self.port
322 }
323
324 /// The `ssh` argv this host is reached with, minus the target: the shared
325 /// [`SSH_FLAGS`] plus `-p <port>` when one is set. Shared by the exec path
326 /// ([`RemoteHost::command`]) and the sync path (rsync's `-e`), so the two
327 /// cannot drift.
328 pub(crate) fn ssh_args(&self) -> Vec<String> {
329 let mut args: Vec<String> = SSH_FLAGS
330 .iter()
331 .map(std::string::ToString::to_string)
332 .collect();
333 if let Some(p) = self.port {
334 args.push("-p".into());
335 args.push(p.to_string());
336 }
337 args
338 }
339
340 /// Build the `Command` that runs `script` as a single `/bin/sh` program,
341 /// locally or over SSH. The remote side runs `script` as the argument to
342 /// the login shell (ssh joins argv with spaces and hands it to the shell),
343 /// so multi-statement scripts and pipes work the same as locally.
344 pub(crate) fn command(&self, script: &str) -> Command {
345 if self.is_local() {
346 let mut cmd = Command::new("sh");
347 cmd.arg("-c").arg(script);
348 cmd
349 } else {
350 let mut cmd = Command::new("ssh");
351 cmd.args(self.ssh_args()).arg(&self.ssh_target).arg(script);
352 cmd
353 }
354 }
355
356 /// Spawn `script`, stream its merged output into `sink`, and return the
357 /// exit status plus captured bytes. `kill_on_drop` means cancelling the
358 /// caller's task (e.g. a newer build superseding this one) SIGKILLs the
359 /// child and — for SSH — drops the connection.
360 /// An SSH host cannot be trusted to report exit status (see [`RcSentinel`]),
361 /// so remote scripts carry their code back in-band. Local `sh -c` reports it
362 /// honestly and needs no sentinel.
363 pub(crate) fn rc_sentinel(&self) -> Option<RcSentinel> {
364 (!self.is_local()).then(RcSentinel::new)
365 }
366
367 /// The `Command` to run `script` on this host, plus the sentinel its output
368 /// will carry (if any). The one place a script is prepared for a host, so
369 /// no execution path can forget to wrap it.
370 pub(crate) fn command_for(&self, script: &str) -> (Command, Option<RcSentinel>) {
371 let sentinel = self.rc_sentinel();
372 let script = match &sentinel {
373 Some(s) => s.wrap(script),
374 None => script.to_string(),
375 };
376 (self.command(&script), sentinel)
377 }
378
379 pub async fn run_streaming<S>(&self, script: &str, sink: Arc<Mutex<S>>) -> Result<RunOutput>
380 where
381 S: LogSink + Send + 'static,
382 {
383 let (mut cmd, sentinel) = self.command_for(script);
384 let mut child = cmd
385 .stdout(Stdio::piped())
386 .stderr(Stdio::piped())
387 .kill_on_drop(true)
388 .spawn()
389 .with_context(|| format!("spawning command on {}", self.ssh_target))?;
390
391 // Only stdout carries the sentinel; stderr streams untouched.
392 let stdout_task = tokio::spawn(drain(
393 child.stdout.take(),
394 sink.clone(),
395 RcFilter::new(sentinel.clone()),
396 ));
397 let stderr_task = tokio::spawn(drain(
398 child.stderr.take(),
399 sink.clone(),
400 RcFilter::new(None),
401 ));
402 let status = child.wait().await.context("waiting on child")?;
403 let (stdout, code) = stdout_task.await.unwrap_or_default();
404 let (stderr, _) = stderr_task.await.unwrap_or_default();
405 let status = match sentinel {
406 Some(_) => resolve_remote_status(&self.ssh_target, status, code)?,
407 None => status,
408 };
409 Ok(RunOutput {
410 status,
411 stdout,
412 stderr,
413 })
414 }
415 }
416
417 /// Drain `stream` into the shared sink and return the concatenated bytes plus
418 /// any exit code the sentinel carried.
419 async fn drain<R, S>(
420 stream: Option<R>,
421 sink: Arc<Mutex<S>>,
422 mut filter: RcFilter,
423 ) -> (Vec<u8>, Option<i32>)
424 where
425 R: AsyncRead + Unpin + Send + 'static,
426 S: LogSink + Send + 'static,
427 {
428 let mut total = Vec::new();
429 let Some(mut s) = stream else {
430 return (total, None);
431 };
432 let mut buf = [0u8; 4096];
433 loop {
434 match s.read(&mut buf).await {
435 Ok(0) | Err(_) => break,
436 Ok(n) => {
437 let out = filter.feed(&buf[..n]);
438 if !out.is_empty() {
439 push_bounded(&mut total, &out, OUTPUT_TAIL_CAP);
440 sink.lock().await.write_chunk(&out).await;
441 }
442 }
443 }
444 }
445 let (rest, code) = filter.finish();
446 if !rest.is_empty() {
447 push_bounded(&mut total, &rest, OUTPUT_TAIL_CAP);
448 sink.lock().await.write_chunk(&rest).await;
449 }
450 (total, code)
451 }
452
453 /// Single-quote a string for safe inclusion in a `/bin/sh` command. This is
454 /// complete POSIX single-quoting: the result is a single-quoted literal with
455 /// every embedded `'` rewritten as `'\''` (close-quote, escaped quote,
456 /// reopen-quote). Inside single quotes the shell treats every other byte —
457 /// `$`, backtick, `;`, `\`, newline — literally, so no metacharacter can act
458 /// and there is no way to break out. Proven adversarially by
459 /// `sh_quote_neutralizes_injection_through_real_sh`.
460 pub fn sh_quote(s: &str) -> String {
461 let escaped = s.replace('\'', r"'\''");
462 format!("'{escaped}'")
463 }
464
465 #[cfg(test)]
466 mod tests {
467 use super::*;
468
469 #[test]
470 fn ssh_args_carry_the_port_and_the_shared_flags() {
471 // The point of the port living on RemoteHost: the exec path and the
472 // sync path build their ssh invocation from this one place, so they
473 // cannot disagree about which port a host is on.
474 let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
475 let args = host.ssh_args();
476 let p = args.iter().position(|a| a == "-p").expect("-p present");
477 assert_eq!(args[p + 1], "2222");
478 assert!(args.contains(&"BatchMode=yes".to_string()));
479
480 // Default: no -p, so ssh/ssh_config picks the port.
481 assert!(
482 !RemoteHost::new("mbp")
483 .ssh_args()
484 .contains(&"-p".to_string())
485 );
486 }
487
488 #[test]
489 fn ported_host_puts_the_port_on_the_ssh_command() {
490 let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
491 let cmd = host.command("true");
492 let args: Vec<String> = cmd
493 .as_std()
494 .get_args()
495 .map(|a| a.to_string_lossy().to_string())
496 .collect();
497 let p = args
498 .iter()
499 .position(|a| a == "-p")
500 .expect("-p on the ssh argv");
501 assert_eq!(args[p + 1], "2222");
502 // The script is still the last arg, after the target.
503 assert_eq!(args.last().unwrap(), "true");
504 }
505
506 #[test]
507 fn local_host_ignores_the_port() {
508 // "local" runs via `sh -c`; a port is meaningless and must not leak in.
509 let host = RemoteHost::new("local").with_port(Some(2222));
510 let cmd = host.command("true");
511 let args: Vec<String> = cmd
512 .as_std()
513 .get_args()
514 .map(|a| a.to_string_lossy().to_string())
515 .collect();
516 assert_eq!(args, vec!["-c".to_string(), "true".to_string()]);
517 }
518
519 #[test]
520 fn push_bounded_keeps_only_the_last_cap_bytes() {
521 let mut buf = Vec::new();
522 push_bounded(&mut buf, b"hello", 8);
523 push_bounded(&mut buf, b"world", 8); // 10 bytes -> trim front to last 8
524 assert_eq!(buf, b"lloworld");
525 // A single chunk larger than the cap is itself trimmed to the tail.
526 let mut buf2 = Vec::new();
527 push_bounded(&mut buf2, b"0123456789", 4);
528 assert_eq!(buf2, b"6789");
529 // Under the cap: untouched.
530 let mut buf3 = Vec::new();
531 push_bounded(&mut buf3, b"ok", 8);
532 assert_eq!(buf3, b"ok");
533 }
534
535 /// A simple sink that accumulates every chunk for assertions.
536 #[derive(Default)]
537 pub(crate) struct VecSink(pub Vec<u8>);
538 #[async_trait]
539 impl LogSink for VecSink {
540 async fn write_chunk(&mut self, bytes: &[u8]) {
541 self.0.extend_from_slice(bytes);
542 }
543 }
544
545 #[test]
546 fn sh_quote_escapes() {
547 assert_eq!(sh_quote("hello"), "'hello'");
548 assert_eq!(sh_quote("it's"), r"'it'\''s'");
549 }
550
551 /// Adversarial proof: run a crafted payload through a real `/bin/sh` and
552 /// confirm it round-trips verbatim — i.e. the shell treated it as a literal
553 /// and no expansion, substitution, or command injection occurred.
554 #[tokio::test]
555 async fn sh_quote_neutralizes_injection_through_real_sh() {
556 use tokio::process::Command;
557 let payloads = [
558 "v'; touch /tmp/ops-exec-should-not-exist; echo '",
559 "$(echo pwned)",
560 "`echo pwned`",
561 "a\\b\\c",
562 "line1\nline2",
563 "$'\\x41'",
564 "'; rm -rf / #",
565 "${HOME}",
566 "* ? [a-z] | & ; ( ) < >",
567 ];
568 for payload in payloads {
569 let cmd = format!("printf '%s' {}", sh_quote(payload));
570 let out = Command::new("sh")
571 .arg("-c")
572 .arg(&cmd)
573 .output()
574 .await
575 .unwrap();
576 assert!(out.status.success(), "sh failed for {payload:?}");
577 assert_eq!(
578 String::from_utf8_lossy(&out.stdout),
579 payload,
580 "sh_quote did not round-trip {payload:?} verbatim (injection or expansion occurred)"
581 );
582 }
583 // The injection side effect must never have happened.
584 assert!(!std::path::Path::new("/tmp/ops-exec-should-not-exist").exists());
585 }
586
587 #[test]
588 fn local_detection() {
589 assert!(RemoteHost::new("local").is_local());
590 assert!(RemoteHost::new("").is_local());
591 assert!(!RemoteHost::new("mbp").is_local());
592 }
593
594 #[tokio::test]
595 async fn local_run_streams_stdout_and_captures_status() {
596 let host = RemoteHost::new("local");
597 let sink = Arc::new(Mutex::new(VecSink::default()));
598 let out = host
599 .run_streaming("printf 'hello '; printf 'world'", sink.clone())
600 .await
601 .unwrap();
602 assert!(out.success());
603 assert_eq!(out.stdout, b"hello world");
604 assert_eq!(sink.lock().await.0, b"hello world");
605 }
606
607 #[tokio::test]
608 async fn local_run_streams_stderr_too() {
609 let host = RemoteHost::new("local");
610 let sink = Arc::new(Mutex::new(VecSink::default()));
611 let out = host
612 .run_streaming("echo out; echo err 1>&2", sink.clone())
613 .await
614 .unwrap();
615 assert!(out.success());
616 assert_eq!(out.stdout, b"out\n");
617 assert_eq!(out.stderr, b"err\n");
618 let seen = sink.lock().await.0.clone();
619 assert!(seen.windows(4).any(|w| w == b"out\n"));
620 assert!(seen.windows(4).any(|w| w == b"err\n"));
621 }
622
623 #[tokio::test]
624 async fn local_run_reports_nonzero_exit() {
625 let host = RemoteHost::new("local");
626 let sink = Arc::new(Mutex::new(VecSink::default()));
627 let out = host.run_streaming("exit 3", sink).await.unwrap();
628 assert!(!out.success());
629 assert_eq!(out.status.code(), Some(3));
630 }
631
632 #[test]
633 fn local_gets_no_sentinel_and_ssh_does() {
634 assert!(RemoteHost::new("local").rc_sentinel().is_none());
635 assert!(RemoteHost::new("mbp").rc_sentinel().is_some());
636 }
637
638 #[test]
639 fn each_sentinel_has_its_own_nonce() {
640 // Build output cannot forge a marker it has never seen.
641 assert_ne!(RcSentinel::new().body, RcSentinel::new().body);
642 }
643
644 /// Drive a wrapped script through a real `/bin/sh` (standing in for the
645 /// remote login shell) and report what the sentinel carried back.
646 async fn run_wrapped(script: &str) -> (Vec<u8>, Option<i32>) {
647 let s = RcSentinel::new();
648 let out = tokio::process::Command::new("sh")
649 .arg("-c")
650 .arg(s.wrap(script))
651 .output()
652 .await
653 .unwrap();
654 let mut filter = RcFilter::new(Some(s));
655 let mut seen = filter.feed(&out.stdout);
656 let (rest, code) = filter.finish();
657 seen.extend_from_slice(&rest);
658 (seen, code)
659 }
660
661 #[tokio::test]
662 async fn sentinel_carries_the_real_code_through_a_shell() {
663 for code in [0, 1, 42, 255] {
664 let (_, got) = run_wrapped(&format!("exit {code}")).await;
665 assert_eq!(
666 got,
667 Some(code),
668 "wrapped `exit {code}` should report {code}"
669 );
670 }
671 }
672
673 /// The reason for the subshell: our callers send `set -e` scripts, and an
674 /// errexit at the top level of the remote shell would kill it before the
675 /// sentinel printed — losing the code on exactly the failures that matter.
676 #[tokio::test]
677 async fn set_e_failure_still_reports_its_code() {
678 let (_, code) = run_wrapped("set -e; false; echo unreachable").await;
679 assert_eq!(code, Some(1));
680 }
681
682 /// Same for an explicit `exit`, which the driver's `set -e; git ...` scripts
683 /// reach via a failing command.
684 #[tokio::test]
685 async fn explicit_exit_still_reports_its_code() {
686 let (out, code) = run_wrapped("echo before; exit 7").await;
687 assert_eq!(code, Some(7));
688 assert_eq!(out, b"before\n");
689 }
690
691 #[tokio::test]
692 async fn the_sentinel_never_reaches_the_caller_or_the_log() {
693 // Output with and without a trailing newline: the wrapper adds a leading
694 // \n so the marker always starts a line, and stripping must put the
695 // stream back exactly as the script wrote it.
696 let (out, code) = run_wrapped("printf 'no trailing newline'").await;
697 assert_eq!(code, Some(0));
698 assert_eq!(String::from_utf8_lossy(&out), "no trailing newline");
699
700 let (out, code) = run_wrapped("echo 'with trailing newline'").await;
701 assert_eq!(code, Some(0));
702 assert_eq!(String::from_utf8_lossy(&out), "with trailing newline\n");
703 }
704
705 /// A build log that prints something sentinel-shaped must not be able to
706 /// pass itself off as the exit code: the nonce is unguessable and the real
707 /// sentinel is always last.
708 #[tokio::test]
709 async fn script_output_cannot_forge_the_code() {
710 let (out, code) = run_wrapped("echo '__ops_exec_rc_deadbeef=0'; exit 9").await;
711 assert_eq!(
712 code,
713 Some(9),
714 "the forged line must not be read as the code"
715 );
716 assert_eq!(String::from_utf8_lossy(&out), "__ops_exec_rc_deadbeef=0\n");
717 }
718
719 /// The filter holds back a tail; a sentinel split across read boundaries
720 /// must still be recognized (and never leak a fragment into the log).
721 #[test]
722 fn sentinel_split_across_chunks_is_still_stripped() {
723 let s = RcSentinel::new();
724 let stream = format!("hello\n\n{}42\n", s.body).into_bytes();
725 let mut filter = RcFilter::new(Some(s));
726 let mut seen = Vec::new();
727 for chunk in stream.chunks(3) {
728 seen.extend_from_slice(&filter.feed(chunk));
729 }
730 let (rest, code) = filter.finish();
731 seen.extend_from_slice(&rest);
732 assert_eq!(code, Some(42));
733 assert_eq!(String::from_utf8_lossy(&seen), "hello\n");
734 }
735
736 #[test]
737 fn a_missing_sentinel_yields_no_code_and_keeps_every_byte() {
738 let mut filter = RcFilter::new(Some(RcSentinel::new()));
739 let mut seen = filter.feed(b"build output, no sentinel\n");
740 let (rest, code) = filter.finish();
741 seen.extend_from_slice(&rest);
742 assert_eq!(code, None);
743 assert_eq!(
744 String::from_utf8_lossy(&seen),
745 "build output, no sentinel\n"
746 );
747 }
748
749 /// The whole point: a transport that claims success without a sentinel is
750 /// reported as an error, never as a green step.
751 #[test]
752 fn success_without_a_sentinel_fails_closed() {
753 let err = resolve_remote_status("mbp", exit_status_from_code(0), None).unwrap_err();
754 let msg = err.to_string();
755 assert!(msg.contains("mbp"), "should name the host: {msg}");
756 assert!(msg.contains("no exit-status sentinel"), "{msg}");
757 }
758
759 #[test]
760 fn the_sentinel_outranks_what_the_transport_claimed() {
761 // Tailscale SSH's lie: ssh says 0, the remote command really failed.
762 let got = resolve_remote_status("mbp", exit_status_from_code(0), Some(42)).unwrap();
763 assert_eq!(got.code(), Some(42));
764 assert!(!got.success());
765
766 // And a real success is still a success.
767 let got = resolve_remote_status("mbp", exit_status_from_code(0), Some(0)).unwrap();
768 assert!(got.success());
769 }
770
771 /// ssh exits 255 and writes no sentinel when it cannot connect. That is a
772 /// real failure with a real code, not the ambiguous case — surface it.
773 #[test]
774 fn a_transport_failure_with_no_sentinel_is_reported_as_itself() {
775 let got = resolve_remote_status("mbp", exit_status_from_code(255), None).unwrap();
776 assert_eq!(got.code(), Some(255));
777 }
778 }
779