fix codex detection, add new security-focused smoketests. (#6682)
Fix 'codex' detection to look for debug build, then release build, then installed. Adds more smoketests around security from @viyatb-oai
This commit is contained in:
iceweasel-oai
GitHub
parent
936650001f
commit
abb7b79701
1 changed files with 178 additions and 42 deletions
|
|
@ -12,36 +12,30 @@ from typing import List, Optional, Tuple
|
|||
def _resolve_codex_cmd() -> List[str]:
|
||||
"""Resolve the Codex CLI to invoke `codex sandbox windows`.
|
||||
|
||||
Prefer `codex` on PATH; if not found, try common local build locations.
|
||||
Prefer local builds (debug first), then fall back to PATH.
|
||||
Returns the argv prefix to run Codex.
|
||||
"""
|
||||
# 1) Prefer PATH
|
||||
try:
|
||||
cp = subprocess.run(["where", "codex"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
|
||||
if cp.returncode == 0:
|
||||
for line in cp.stdout.splitlines():
|
||||
p = Path(line.strip())
|
||||
if p.exists():
|
||||
return [str(p)]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2) Try workspace targets
|
||||
root = Path(__file__).parent
|
||||
ws_root = root.parent
|
||||
cargo_target = os.environ.get("CARGO_TARGET_DIR")
|
||||
|
||||
candidates = [
|
||||
ws_root / "target" / "release" / "codex.exe",
|
||||
ws_root / "target" / "debug" / "codex.exe",
|
||||
ws_root / "target" / "release" / "codex.exe",
|
||||
]
|
||||
if cargo_target:
|
||||
cargo_base = Path(cargo_target)
|
||||
candidates.extend([
|
||||
Path(cargo_target) / "release" / "codex.exe",
|
||||
Path(cargo_target) / "debug" / "codex.exe",
|
||||
cargo_base / "debug" / "codex.exe",
|
||||
cargo_base / "release" / "codex.exe",
|
||||
])
|
||||
for p in candidates:
|
||||
if p.exists():
|
||||
return [str(p)]
|
||||
|
||||
for candidate in candidates:
|
||||
if candidate.exists():
|
||||
return [str(candidate)]
|
||||
|
||||
if shutil.which("codex"):
|
||||
return ["codex"]
|
||||
|
||||
raise FileNotFoundError(
|
||||
"Codex CLI not found. Build it first, e.g.\n"
|
||||
|
|
@ -51,6 +45,7 @@ def _resolve_codex_cmd() -> List[str]:
|
|||
)
|
||||
|
||||
CODEX_CMD = _resolve_codex_cmd()
|
||||
print(CODEX_CMD)
|
||||
TIMEOUT_SEC = 20
|
||||
|
||||
WS_ROOT = Path(os.environ["USERPROFILE"]) / "sbx_ws_tests"
|
||||
|
|
@ -81,11 +76,7 @@ def run_sbx(policy: str, cmd_argv: List[str], cwd: Path, env_extra: Optional[dic
|
|||
return cp.returncode, cp.stdout, cp.stderr
|
||||
|
||||
def have(cmd: str) -> bool:
|
||||
try:
|
||||
cp = subprocess.run(["where", cmd], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
|
||||
return cp.returncode == 0 and any(Path(p.strip()).exists() for p in cp.stdout.splitlines())
|
||||
except Exception:
|
||||
return False
|
||||
return shutil.which(cmd) is not None
|
||||
|
||||
def make_dir_clean(p: Path) -> None:
|
||||
if p.exists():
|
||||
|
|
@ -109,6 +100,26 @@ def assert_exists(p: Path) -> bool:
|
|||
def assert_not_exists(p: Path) -> bool:
|
||||
return not p.exists()
|
||||
|
||||
def make_junction(link: Path, target: Path) -> bool:
|
||||
"""Create a directory junction; return True if it exists afterward."""
|
||||
remove_if_exists(link)
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
cmd = ["cmd", "/c", f'mklink /J "{link}" "{target}"']
|
||||
cp = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||||
return cp.returncode == 0 and link.exists()
|
||||
|
||||
def make_symlink(link: Path, target: Path) -> bool:
|
||||
"""Create a directory symlink; return True if it exists afterward."""
|
||||
remove_if_exists(link)
|
||||
if not target.exists():
|
||||
try:
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
cmd = ["cmd", "/c", f'mklink /D "{link}" "{target}"']
|
||||
cp = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||||
return cp.returncode == 0 and link.exists()
|
||||
|
||||
def summarize(results: List[CaseResult]) -> int:
|
||||
ok = sum(1 for r in results if r.ok)
|
||||
total = len(results)
|
||||
|
|
@ -255,24 +266,6 @@ def main() -> int:
|
|||
else:
|
||||
add("WS: git --version (optional, skipped)", True)
|
||||
|
||||
# 21–23. JSON policy: allow only .\allowed — note CWD is still allowed by current impl
|
||||
(WS_ROOT / "allowed").mkdir(exist_ok=True)
|
||||
(WS_ROOT / "blocked").mkdir(exist_ok=True)
|
||||
policy_json = '{"mode":"workspace-write","workspace_roots":[".\\\\allowed"]}'
|
||||
|
||||
# Allowed: inside .\allowed (OK)
|
||||
rc, out, err = run_sbx(policy_json, ["cmd", "/c", "echo ok > allowed\\in_allowed.txt"], WS_ROOT)
|
||||
add("JSON WS: write in allowed/ OK", rc == 0 and (WS_ROOT / "allowed/in_allowed.txt").exists(), f"rc={rc}")
|
||||
|
||||
# Outside CWD (deny)
|
||||
json_outside = OUTSIDE / "json_blocked.txt"; remove_if_exists(json_outside)
|
||||
rc, out, err = run_sbx(policy_json, ["cmd", "/c", f"echo nope > {json_outside}"], WS_ROOT)
|
||||
add("JSON WS: write outside allowed/ denied", rc != 0 and not json_outside.exists(), f"rc={rc}")
|
||||
|
||||
# CWD is still allowed by current sandbox (documented behavior)
|
||||
rc, out, err = run_sbx(policy_json, ["cmd", "/c", "echo ok > cwd_ok_under_json.txt"], WS_ROOT)
|
||||
add("JSON WS: write in CWD allowed (by design)", rc == 0 and (WS_ROOT / "cwd_ok_under_json.txt").exists(), f"rc={rc}")
|
||||
|
||||
# 24. WS: PS bytes write (OK)
|
||||
rc, out, err = run_sbx("workspace-write",
|
||||
["powershell", "-NoLogo", "-NoProfile", "-Command",
|
||||
|
|
@ -300,6 +293,149 @@ def main() -> int:
|
|||
rc, out, err = run_sbx("read-only", ["cmd", "/c", "echo nope > cmd_ro.txt"], WS_ROOT)
|
||||
add("RO: cmd redirection denied", rc != 0 and not target.exists(), f"rc={rc}")
|
||||
|
||||
# 29. WS: CWD junction poisoning denied (allowlist should not follow to OUTSIDE)
|
||||
poison_cwd = WS_ROOT / "poison_cwd"
|
||||
if make_junction(poison_cwd, OUTSIDE):
|
||||
target = OUTSIDE / "poisoned.txt"
|
||||
remove_if_exists(target)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo poison > poisoned.txt"], poison_cwd)
|
||||
add("WS: junction poisoning via CWD denied", rc != 0 and assert_not_exists(target), f"rc={rc}, err={err}")
|
||||
else:
|
||||
add("WS: junction poisoning via CWD denied (setup skipped)", True, "junction creation failed")
|
||||
|
||||
# 30. WS: junction into Windows denied
|
||||
sys_link = WS_ROOT / "sys_link"
|
||||
sys_target = Path("C:/Windows")
|
||||
sys_file = sys_target / "system32" / "sbx_junc.txt"
|
||||
if sys_file.exists():
|
||||
remove_if_exists(sys_file)
|
||||
if make_junction(sys_link, sys_target):
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo bad > sys_link\\system32\\sbx_junc.txt"], WS_ROOT)
|
||||
add("WS: junction into Windows denied", rc != 0 and not sys_file.exists(), f"rc={rc}, err={err}")
|
||||
else:
|
||||
add("WS: junction into Windows denied (setup skipped)", True, "junction creation failed")
|
||||
|
||||
# 31. WS: device/pipe access blocked
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "type \\\\.\\PhysicalDrive0"], WS_ROOT)
|
||||
add("WS: raw device access denied", rc != 0, f"rc={rc}")
|
||||
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo hi > \\\\.\\pipe\\codex_testpipe"], WS_ROOT)
|
||||
add("WS: named pipe creation denied", rc != 0, f"rc={rc}")
|
||||
|
||||
# 32. WS: ADS/long-path escape denied
|
||||
ads_base = WS_ROOT / "ads_base.txt"
|
||||
remove_if_exists(ads_base)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo secret > ads_base.txt:stream"], WS_ROOT)
|
||||
add("WS: ADS write denied", rc != 0 and assert_not_exists(ads_base), f"rc={rc}")
|
||||
|
||||
lp_target = Path(r"\\?\C:\sbx_longpath_test.txt")
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo long > \\\\?\\C:\\sbx_longpath_test.txt"], WS_ROOT)
|
||||
add("WS: long-path escape denied", rc != 0 and not lp_target.exists(), f"rc={rc}")
|
||||
|
||||
# 33. WS: case-insensitive protected path bypass denied (.GiT)
|
||||
git_variation = WS_ROOT / ".GiT" / "config"
|
||||
remove_if_exists(git_variation.parent)
|
||||
git_variation.parent.mkdir(exist_ok=True)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo hack > .GiT\\config"], WS_ROOT)
|
||||
add("WS: protected path case-variation denied", rc != 0 and assert_not_exists(git_variation), f"rc={rc}")
|
||||
|
||||
# 34. WS: policy tamper (.codex artifacts) denied
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo tamper > .codex\\cap_sid"], WS_ROOT)
|
||||
rc2, out2, err2 = run_sbx("workspace-write", ["cmd", "/c", "echo tamper > .codex\\policy.json"], WS_ROOT)
|
||||
add("WS: .codex cap_sid tamper denied", rc != 0, f"rc={rc}, err={err}")
|
||||
add("WS: .codex policy tamper denied", rc2 != 0, f"rc={rc2}, err={err2}")
|
||||
|
||||
# 35. WS: PATH stub bypass denied (ssh before stubs)
|
||||
tools_dir = WS_ROOT / "tools"
|
||||
tools_dir.mkdir(exist_ok=True)
|
||||
ssh_path = None
|
||||
if have("ssh"):
|
||||
# shutil.which considers PATHEXT + PATHEXT semantics
|
||||
ssh_path = shutil.which("ssh")
|
||||
if ssh_path:
|
||||
shim = tools_dir / "ssh.bat"
|
||||
shim.write_text("@echo off\r\necho stubbed\r\n", encoding="utf-8")
|
||||
env = {"PATH": f"{tools_dir};%PATH%"}
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "ssh"], WS_ROOT, env_extra=env)
|
||||
add("WS: PATH stub bypass denied", "stubbed" in out, f"rc={rc}, out={out}")
|
||||
else:
|
||||
add("WS: PATH stub bypass denied (ssh missing)", True, "ssh not installed")
|
||||
|
||||
# 36. WS: symlink races blocked
|
||||
race_root = WS_ROOT / "race"
|
||||
inside = race_root / "inside"
|
||||
outside = race_root / "outside"
|
||||
make_dir_clean(race_root)
|
||||
inside.mkdir(parents=True, exist_ok=True)
|
||||
outside.mkdir(parents=True, exist_ok=True)
|
||||
link = race_root / "flip"
|
||||
make_symlink(link, inside)
|
||||
# Fire a quick toggle loop and attempt a write
|
||||
outside_abs = str(OUTSIDE)
|
||||
inside_abs = str(inside)
|
||||
toggle = [
|
||||
"cmd",
|
||||
"/c",
|
||||
f'for /L %i in (1,1,400) do (rmdir flip & mklink /D flip "{inside_abs}" >NUL & rmdir flip & mklink /D flip "{outside_abs}" >NUL)',
|
||||
]
|
||||
subprocess.Popen(toggle, cwd=str(race_root), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo race > flip\\race.txt"], race_root)
|
||||
add("WS: symlink race write denied (best-effort)", rc != 0 and not (outside / "race.txt").exists(), f"rc={rc}")
|
||||
|
||||
# 37. WS: audit blind spots – deep junction/world-writable denied
|
||||
deep = WS_ROOT / "deep" / "redir"
|
||||
unsafe_dir = WS_ROOT / "deep" / "unsafe"
|
||||
make_junction(deep, Path("C:/Windows"))
|
||||
unsafe_dir.mkdir(parents=True, exist_ok=True)
|
||||
subprocess.run(["icacls", str(unsafe_dir), "/grant", "Everyone:(F)"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo probe > deep\\redir\\system32\\audit_gap.txt"], WS_ROOT)
|
||||
add("WS: deep junction/world-writable escape denied", rc != 0, f"rc={rc}, err={err}")
|
||||
|
||||
# 38. WS: policy poisoning via workspace symlink root denied
|
||||
# Simulate workspace replaced by symlink to C:\; expect writes to be denied.
|
||||
fake_root = WS_ROOT / "fake_root"
|
||||
if make_symlink(fake_root, Path("C:/")):
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo owned > codex_escape.txt"], fake_root)
|
||||
add("WS: workspace-root symlink poisoning denied", rc != 0, f"rc={rc}")
|
||||
else:
|
||||
add("WS: workspace-root symlink poisoning denied (setup skipped)", True, "symlink creation failed")
|
||||
|
||||
# 39. WS: UNC/other-drive canonicalization denied
|
||||
unc_link = WS_ROOT / "unc_link"
|
||||
other_to = Path(r"\\\\localhost\\C$")
|
||||
if make_symlink(unc_link, other_to):
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo unc > unc_link\\unc_test.txt"], WS_ROOT)
|
||||
add("WS: UNC link escape denied", rc != 0, f"rc={rc}")
|
||||
else:
|
||||
add("WS: UNC link escape denied (setup skipped)", True, "symlink creation failed")
|
||||
|
||||
other_drive = WS_ROOT / "other_drive"
|
||||
other_target = Path("D:/") # best-effort; may not exist
|
||||
if make_symlink(other_drive, other_target):
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo drive > other_drive\\drive.txt"], WS_ROOT)
|
||||
add("WS: other-drive link escape denied", rc != 0, f"rc={rc}")
|
||||
else:
|
||||
add("WS: other-drive link escape denied (setup skipped)", True, "symlink creation failed")
|
||||
|
||||
# 40. WS: timeout cleanup still denies outside write
|
||||
slow_ps = WS_ROOT / "sleep.ps1"
|
||||
slow_ps.write_text("Start-Sleep 15", encoding="utf-8")
|
||||
try:
|
||||
run_sbx("workspace-write", ["powershell", "-File", "sleep.ps1"], WS_ROOT)
|
||||
except Exception:
|
||||
pass
|
||||
outside_after_timeout = OUTSIDE / "timeout_leak.txt"
|
||||
remove_if_exists(outside_after_timeout)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", f"echo leak > {outside_after_timeout}"], WS_ROOT)
|
||||
add("WS: post-timeout outside write still denied", rc != 0 and assert_not_exists(outside_after_timeout), f"rc={rc}")
|
||||
|
||||
# 41. WS: additional protected path variation (.ssh)
|
||||
ssh_var = WS_ROOT / ".SsH" / "config"
|
||||
remove_if_exists(ssh_var.parent)
|
||||
ssh_var.parent.mkdir(exist_ok=True)
|
||||
rc, out, err = run_sbx("workspace-write", ["cmd", "/c", "echo key > .SsH\\config"], WS_ROOT)
|
||||
add("WS: protected path variation (.ssh) denied", rc != 0 and assert_not_exists(ssh_var), f"rc={rc}")
|
||||
|
||||
return summarize(results)
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue