mirror of
https://github.com/microsoft/SkillOpt.git
synced 2026-08-03 07:02:46 +08:00
fix(spreadsheet): complete generated-code env isolation
This commit is contained in:
@@ -262,6 +262,7 @@ def _build_codex_driver() -> str:
|
||||
"import re\n"
|
||||
"import subprocess\n"
|
||||
"import sys\n\n"
|
||||
"import tempfile\n\n"
|
||||
'INPUT_PATH = "input.xlsx"\n'
|
||||
'OUTPUT_PATH = "output.xlsx"\n'
|
||||
"code = pathlib.Path('solution.py').read_text(encoding='utf-8')\n"
|
||||
@@ -277,15 +278,21 @@ def _build_codex_driver() -> str:
|
||||
"_safe_env = {\n"
|
||||
" 'PATH': _os.environ.get('PATH', '/usr/bin:/bin'),\n"
|
||||
" 'HOME': str(pathlib.Path('_driver_runner.py').parent.resolve()),\n"
|
||||
" 'TMPDIR': tempfile.gettempdir(),\n"
|
||||
"}\n"
|
||||
"if _os.name == 'nt':\n"
|
||||
" _safe_env['SYSTEMROOT'] = _os.environ.get('SYSTEMROOT', '')\n"
|
||||
" _safe_env['TEMP'] = _safe_env['HOME']\n"
|
||||
" _safe_env['TMP'] = _safe_env['HOME']\n"
|
||||
" _safe_env['TEMP'] = tempfile.gettempdir()\n"
|
||||
" _safe_env['TMP'] = tempfile.gettempdir()\n"
|
||||
"_safe_env = {k: v for k, v in _safe_env.items() if v}\n"
|
||||
"_res = subprocess.run([sys.executable, str(_patched)], capture_output=True, text=True, env=_safe_env)\n"
|
||||
"try:\n"
|
||||
" _res = subprocess.run([sys.executable, str(_patched)], capture_output=True, text=True, env=_safe_env)\n"
|
||||
"finally:\n"
|
||||
" try:\n"
|
||||
" _patched.unlink()\n"
|
||||
" except OSError:\n"
|
||||
" pass\n"
|
||||
"if _res.returncode != 0:\n"
|
||||
" import traceback as _tb\n"
|
||||
" print(_res.stdout, end='')\n"
|
||||
" print(_res.stderr, end='')\n"
|
||||
" sys.exit(2)\n"
|
||||
|
||||
@@ -46,8 +46,9 @@ def run_generated_code(code: str, input_path: str, output_path: str, timeout: in
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
|
||||
f.write(script)
|
||||
tmp = f.name
|
||||
# Build a minimal environment so the generated code cannot read API keys,
|
||||
# cloud credentials, or other secrets from the current process environment.
|
||||
# Build a minimal environment so generated code does not directly inherit
|
||||
# API keys, cloud credentials, or other parent-process environment values.
|
||||
# This is environment isolation, not a filesystem or network sandbox.
|
||||
import platform as _platform
|
||||
_safe_env: dict[str, str] = {
|
||||
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
|
||||
@@ -58,7 +59,7 @@ def run_generated_code(code: str, input_path: str, output_path: str, timeout: in
|
||||
_safe_env["SYSTEMROOT"] = os.environ.get("SYSTEMROOT", "")
|
||||
_safe_env["TEMP"] = tempfile.gettempdir()
|
||||
_safe_env["TMP"] = tempfile.gettempdir()
|
||||
# Drop empty entries (env dict values must be non-empty strings)
|
||||
# Omit platform-specific entries that are absent in the parent.
|
||||
_safe_env = {k: v for k, v in _safe_env.items() if v}
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
|
||||
@@ -7,8 +7,10 @@ parent process environment. These tests assert that scrubbing behaviour.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from skillopt.envs.spreadsheetbench.codegen_agent import _build_codex_driver
|
||||
from skillopt.envs.spreadsheetbench.executor import run_generated_code
|
||||
|
||||
|
||||
@@ -44,3 +46,32 @@ def test_path_still_available_to_generated_code(tmp_path) -> None:
|
||||
|
||||
assert ok, err
|
||||
assert out.read_text(encoding="utf-8") == "YES"
|
||||
|
||||
|
||||
def test_codex_driver_scrubs_env_sets_tempdir_and_cleans_runner(
|
||||
tmp_path, monkeypatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("SUPER_SECRET_TOKEN", "do-not-inherit")
|
||||
(tmp_path / "solution.py").write_text(
|
||||
"import os\n"
|
||||
"with open(OUTPUT_PATH, 'w', encoding='utf-8') as f:\n"
|
||||
" f.write('|'.join([\n"
|
||||
" os.environ.get('SUPER_SECRET_TOKEN', 'ABSENT'),\n"
|
||||
" 'TMPDIR' if os.environ.get('TMPDIR') else 'NO_TMPDIR',\n"
|
||||
" ]))\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
driver = tmp_path / "run_solution.py"
|
||||
driver.write_text(_build_codex_driver(), encoding="utf-8")
|
||||
|
||||
proc = subprocess.run(
|
||||
[sys.executable, str(driver)],
|
||||
cwd=tmp_path,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
assert proc.returncode == 0, proc.stdout + proc.stderr
|
||||
assert (tmp_path / "output.xlsx").read_text(encoding="utf-8") == "ABSENT|TMPDIR"
|
||||
assert not (tmp_path / "_driver_runner.py").exists()
|
||||
|
||||
Reference in New Issue
Block a user