File size: 5,927 Bytes
adaffb2 | 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 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """Bounded subprocess tests for generated SpreadsheetBench scripts."""
from __future__ import annotations
import os
from pathlib import Path
import subprocess
import time
import openpyxl
from spreadsheetbench_env.executor import ExecutorLimits, ScriptExecutor
def _input_workbook(path: Path) -> None:
workbook = openpyxl.Workbook()
workbook.active["A1"] = 4
workbook.save(path)
workbook.close()
def test_executor_injects_private_paths_and_scrubs_parent_credentials(
tmp_path: Path,
monkeypatch,
) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
monkeypatch.setenv("SPREADSHEETBENCH_TEST_SECRET", "must-not-leak")
executor = ScriptExecutor(tmp_path / "session")
code = """
import os
import openpyxl
INPUT_PATH = '/tmp/attacker-input.xlsx'
OUTPUT_PATH = '/tmp/attacker-output.xlsx'
wb = openpyxl.load_workbook(INPUT_PATH)
wb.active['A2'] = os.environ.get('SPREADSHEETBENCH_TEST_SECRET', 'scrubbed')
wb.active['A3'] = os.environ.get('OPENBLAS_NUM_THREADS')
wb.save(OUTPUT_PATH)
"""
result = executor.run(code, source, case_name="case-1")
assert result.ok is True
assert result.output_path is not None
assert result.output_path.parent == executor.root / "case-1"
output = openpyxl.load_workbook(result.output_path, data_only=True)
assert output.active["A1"].value == 4
assert output.active["A2"].value == "scrubbed"
assert output.active["A3"].value == "1"
output.close()
assert not Path("/tmp/attacker-output.xlsx").exists()
executor.close()
assert not executor.root.exists()
def test_executor_uses_a_distinct_directory_for_each_case(tmp_path: Path) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
executor = ScriptExecutor(tmp_path / "session")
code = "import shutil\nshutil.copyfile(INPUT_PATH, OUTPUT_PATH)\n"
first = executor.run(code, source, case_name="case-1")
second = executor.run(code, source, case_name="case-2")
assert first.ok and second.ok
assert first.output_path != second.output_path
assert first.output_path is not None and second.output_path is not None
assert first.output_path.parent != second.output_path.parent
executor.close()
def test_executor_does_not_use_preexec_fn_in_threaded_server(
tmp_path: Path,
monkeypatch,
) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
executor = ScriptExecutor(tmp_path / "session")
real_popen = subprocess.Popen
received_preexec: list[object] = []
def recording_popen(*args, **kwargs):
received_preexec.append(kwargs.get("preexec_fn"))
return real_popen(*args, **kwargs)
monkeypatch.setattr(subprocess, "Popen", recording_popen)
result = executor.run(
"import shutil\nshutil.copyfile(INPUT_PATH, OUTPUT_PATH)\n",
source,
case_name="no-preexec",
)
assert result.ok is True
assert received_preexec == [None]
executor.close()
def test_timeout_kills_the_generated_process_group(tmp_path: Path) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
executor = ScriptExecutor(
tmp_path / "session",
limits=ExecutorLimits(timeout_s=0.5, cpu_seconds=5),
)
code = """
import subprocess
import sys
import time
child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])
print(child.pid, flush=True)
time.sleep(60)
"""
result = executor.run(code, source, case_name="timeout")
assert result.ok is False
assert result.timed_out is True
assert "timeout" in result.error.lower()
child_pid = int(result.stdout.strip().splitlines()[0])
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
try:
os.kill(child_pid, 0)
except ProcessLookupError:
break
time.sleep(0.05)
else:
raise AssertionError(f"generated child process {child_pid} survived timeout")
executor.close()
def test_timeout_does_not_wait_for_detached_child_output_pipes(tmp_path: Path) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
executor = ScriptExecutor(
tmp_path / "session",
limits=ExecutorLimits(timeout_s=0.2, cpu_seconds=5),
)
code = """
import subprocess
import sys
import time
child = subprocess.Popen(
[sys.executable, '-c', 'import time; time.sleep(10)'],
start_new_session=True,
)
print(child.pid, flush=True)
time.sleep(10)
"""
started = time.monotonic()
result = executor.run(code, source, case_name="detached-child")
elapsed = time.monotonic() - started
assert result.ok is False
assert result.timed_out is True
assert elapsed < 2.0
child_pid = int(result.stdout.strip().splitlines()[0])
try:
os.kill(child_pid, 0)
except ProcessLookupError:
pass
else:
os.kill(child_pid, 9)
executor.close()
def test_executor_rejects_unsafe_case_names(tmp_path: Path) -> None:
source = tmp_path / "source.xlsx"
_input_workbook(source)
executor = ScriptExecutor(tmp_path / "session")
try:
executor.run("", source, case_name="../escape")
except ValueError as error:
assert "case name" in str(error)
else:
raise AssertionError("unsafe case name was accepted")
finally:
executor.close()
def test_concurrent_executors_lease_distinct_runner_uids_and_release_them(
tmp_path: Path,
) -> None:
limits = ExecutorLimits(run_uid_base=20_000, run_gid=65_534)
first = ScriptExecutor(tmp_path / "first", limits=limits)
second = ScriptExecutor(tmp_path / "second", limits=limits)
assert first.leased_uid == 20_000
assert second.leased_uid == 20_001
first.close()
replacement = ScriptExecutor(tmp_path / "replacement", limits=limits)
assert replacement.leased_uid == 20_000
second.close()
replacement.close()
|