File size: 4,021 Bytes
afa0cbf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | //! Exercises the public macOS sandbox runner with a disposable controlling terminal.
#![cfg(target_os = "macos")]
use std::collections::HashMap;
use std::path::Path;
use std::process::Command;
use std::time::Duration;
use anyhow::Context as _;
use codex_utils_pty::SpawnedProcess;
use codex_utils_pty::TerminalSize;
use codex_utils_pty::spawn_pty_process;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
const PROBE: &str = r#"
import errno, fcntl, os, select, signal, struct, sys, termios, tty
signal.alarm(15)
assert os.isatty(0)
assert os.tcgetpgrp(0) == os.getpgrp()
assert struct.unpack("HHHH", fcntl.ioctl(0, termios.TIOCGWINSZ, b"\0" * 8))[:2] == (24, 80)
original = termios.tcgetattr(0)
try:
tty.setraw(0)
print("__ready__", flush=True)
assert os.read(0, 1) == b"k"
try:
fcntl.ioctl(0, termios.TIOCSTI, b"x")
except OSError as error:
assert sys.argv[1] == "deny", error
assert error.errno == errno.EPERM, error
assert not select.select([0], [], [], 0)[0]
else:
assert os.read(0, 1) == b"x"
assert sys.argv[1] == "allow", "TIOCSTI unexpectedly succeeded"
finally:
termios.tcsetattr(0, termios.TCSANOW, original)
print("__passed__", flush=True)
"#;
#[tokio::test]
async fn sandbox_blocks_terminal_input_injection() -> anyhow::Result<()> {
let availability = Command::new("/usr/bin/sandbox-exec")
.args(["-p", "(version 1)(allow default)", "/usr/bin/true"])
.output()?;
if !availability.status.success()
&& String::from_utf8_lossy(&availability.stderr)
.contains("sandbox-exec: sandbox_apply: Operation not permitted")
{
eprintln!("skipping terminal injection test: nested Seatbelt is unavailable");
return Ok(());
}
assert!(availability.status.success(), "{availability:?}");
let codex_home = TempDir::new()?;
let mut env: HashMap<String, String> = std::env::vars().collect();
env.insert(
"CODEX_HOME".to_string(),
codex_home.path().to_string_lossy().into_owned(),
);
// The control proves this is a readable, foreground controlling terminal on
// which TIOCSTI would otherwise work. All injected input stays in this PTY.
run_probe(
"/usr/bin/python3",
vec!["-c".to_string(), PROBE.to_string(), "allow".to_string()],
codex_home.path(),
&env,
)
.await?;
let codex = codex_utils_cargo_bin::cargo_bin("codex")?;
run_probe(
&codex.to_string_lossy(),
vec![
"sandbox".to_string(),
"-P".to_string(),
":read-only".to_string(),
"--".to_string(),
"/usr/bin/python3".to_string(),
"-c".to_string(),
PROBE.to_string(),
"deny".to_string(),
],
codex_home.path(),
&env,
)
.await
}
async fn run_probe(
program: &str,
args: Vec<String>,
cwd: &Path,
env: &HashMap<String, String>,
) -> anyhow::Result<()> {
let SpawnedProcess {
session,
mut stdout_rx,
exit_rx,
..
} = spawn_pty_process(
program,
&args,
cwd,
env,
/*arg0*/ &None,
TerminalSize::default(),
&[],
)
.await?;
let mut output = Vec::new();
let mut sent_input = false;
let code = tokio::time::timeout(Duration::from_secs(30), async {
while let Some(bytes) = stdout_rx.recv().await {
output.extend_from_slice(&bytes);
if !sent_input && String::from_utf8_lossy(&output).contains("__ready__") {
session.writer_sender().send(b"k".to_vec()).await?;
sent_input = true;
}
}
Ok::<_, anyhow::Error>(exit_rx.await?)
})
.await
.context("terminal probe timed out")??;
let output = String::from_utf8_lossy(&output);
assert_eq!(code, 0, "terminal probe failed: {output}");
assert!(output.contains("__passed__"), "{output}");
Ok(())
}
|