fix: harden generated-code isolation and secret redaction

skillopt/envs/spreadsheetbench/codegen_agent.py

@@ -258,40 +258,57 @@ def _build_codex_task(
 
 def _build_codex_driver() -> str:
     return (
+        "import os\n"
         "import pathlib\n"
         "import re\n"
+        "import shutil\n"
         "import subprocess\n"
-        "import sys\n\n"
+        "import sys\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"
         "code = re.sub(r'^\\s*(INPUT_PATH|OUTPUT_PATH)\\s*=\\s*.+$', '', code, flags=re.MULTILINE)\n"
         "# Write patched code to a temporary file and run it in a clean subprocess\n"
-        "# (avoids exec/compile of untrusted LLM-generated code in the current process).\n"
-        "_patched = pathlib.Path('_driver_runner.py')\n"
-        "_patched.write_text(\n"
-        "    f'INPUT_PATH = {INPUT_PATH!r}\\nOUTPUT_PATH = {OUTPUT_PATH!r}\\n' + code,\n"
-        "    encoding='utf-8',\n"
-        ")\n"
-        "import os as _os\n"
-        "_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'] = tempfile.gettempdir()\n"
-        "    _safe_env['TMP'] = tempfile.gettempdir()\n"
-        "_safe_env = {k: v for k, v in _safe_env.items() if v}\n"
+        "# with a scrubbed environment. This avoids in-process exec/compile but is\n"
+        "# not a filesystem, process, or network sandbox.\n"
+        "_work_dir = str(pathlib.Path.cwd())\n"
+        "_temp_dir = tempfile.mkdtemp(prefix='skillopt-generated-')\n"
         "try:\n"
+        "    _patched = pathlib.Path(_temp_dir) / 'runner.py'\n"
+        "    _safe_env = {\n"
+        "        'PATH': os.environ.get('PATH') or os.defpath,\n"
+        "        'HOME': _work_dir,\n"
+        "        'TMPDIR': _temp_dir,\n"
+        "    }\n"
+        "    for _key in (\n"
+        "        'PYTHONPATH', 'PYTHONHOME', 'VIRTUAL_ENV',\n"
+        "        'LD_LIBRARY_PATH', 'DYLD_LIBRARY_PATH',\n"
+        "        'LANG', 'LANGUAGE', 'LC_ALL', 'LC_CTYPE',\n"
+        "        'PYTHONIOENCODING', 'PYTHONUTF8',\n"
+        "        'SYSTEMDRIVE', 'PATHEXT', 'COMSPEC',\n"
+        "    ):\n"
+        "        if os.environ.get(_key):\n"
+        "            _safe_env[_key] = os.environ[_key]\n"
+        "    if os.name == 'nt':\n"
+        "        _safe_env['SYSTEMROOT'] = (\n"
+        "            os.environ.get('SYSTEMROOT')\n"
+        "            or os.environ.get('SystemRoot')\n"
+        "            or os.environ.get('WINDIR', '')\n"
+        "        )\n"
+        "        _safe_env['USERPROFILE'] = _work_dir\n"
+        "        _safe_env['TEMP'] = _temp_dir\n"
+        "        _safe_env['TMP'] = _temp_dir\n"
+        "        _safe_env['APPDATA'] = _temp_dir\n"
+        "        _safe_env['LOCALAPPDATA'] = _temp_dir\n"
+        "    _safe_env = {k: v for k, v in _safe_env.items() if v}\n"
+        "    _patched.write_text(\n"
+        "        f'INPUT_PATH = {INPUT_PATH!r}\\nOUTPUT_PATH = {OUTPUT_PATH!r}\\n' + code,\n"
+        "        encoding='utf-8',\n"
+        "    )\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"
+        "    shutil.rmtree(_temp_dir, ignore_errors=True)\n"
         "if _res.returncode != 0:\n"
         "    print(_res.stdout, end='')\n"
         "    print(_res.stderr, end='')\n"

skillopt/envs/spreadsheetbench/executor.py

@@ -28,56 +28,99 @@
     r'^\s*(INPUT_PATH|OUTPUT_PATH)\s*=\s*.+$', re.MULTILINE
 )
 
+_GENERATED_CODE_ENV_PASSTHROUGH = (
+    # Preserve interpreter/import behavior without inheriting API/cloud keys.
+    "PYTHONPATH",
+    "PYTHONHOME",
+    "VIRTUAL_ENV",
+    "LD_LIBRARY_PATH",
+    "DYLD_LIBRARY_PATH",
+    # Keep text I/O deterministic for non-ASCII spreadsheet content.
+    "LANG",
+    "LANGUAGE",
+    "LC_ALL",
+    "LC_CTYPE",
+    "PYTHONIOENCODING",
+    "PYTHONUTF8",
+    # Needed by executable lookup and CPython on some Windows installations.
+    "SYSTEMDRIVE",
+    "PATHEXT",
+    "COMSPEC",
+)
+
 
 def _strip_path_assignments(code: str) -> str:
     """Remove INPUT_PATH/OUTPUT_PATH assignments from user code."""
     return _PATH_ASSIGN_RE.sub("", code)
 
 
+def generated_code_env(work_dir: str, temp_dir: str) -> dict[str, str]:
+    """Return the minimal environment for LLM-generated spreadsheet Python.
+
+    This prevents direct inheritance of parent-process credentials. It is not a
+    filesystem, process, or network sandbox.
+    """
+    private_dir = os.path.abspath(work_dir or os.getcwd())
+    private_temp = os.path.abspath(temp_dir)
+    safe_env = {
+        "PATH": os.environ.get("PATH") or os.defpath,
+        "HOME": private_dir,
+        "TMPDIR": private_temp,
+    }
+    for key in _GENERATED_CODE_ENV_PASSTHROUGH:
+        value = os.environ.get(key)
+        if value:
+            safe_env[key] = value
+    if os.name == "nt":
+        system_root = (
+            os.environ.get("SYSTEMROOT")
+            or os.environ.get("SystemRoot")
+            or os.environ.get("WINDIR")
+            or ""
+        )
+        safe_env.update({
+            "SYSTEMROOT": system_root,
+            "USERPROFILE": private_dir,
+            "TEMP": private_temp,
+            "TMP": private_temp,
+            "APPDATA": private_temp,
+            "LOCALAPPDATA": private_temp,
+        })
+    return {key: value for key, value in safe_env.items() if value}
+
+
 def run_generated_code(code: str, input_path: str, output_path: str, timeout: int | None = 120) -> tuple[bool, str]:
-    os.makedirs(os.path.dirname(output_path), exist_ok=True)
+    output_dir = os.path.dirname(os.path.abspath(output_path))
+    os.makedirs(output_dir, exist_ok=True)
     cleaned = _strip_path_assignments(code)
     indented = textwrap.indent(cleaned, "    ")
     script = RUNNER_TEMPLATE.format(
         input_path=input_path,
         output_path=output_path,
         user_code_indented=indented,
     )
-    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
-        f.write(script)
-        tmp = f.name
-    # 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"),
-        "HOME": os.path.dirname(output_path),
-        "TMPDIR": tempfile.gettempdir(),
-    }
-    if _platform.system() == "Windows":
-        _safe_env["SYSTEMROOT"] = os.environ.get("SYSTEMROOT", "")
-        _safe_env["TEMP"] = tempfile.gettempdir()
-        _safe_env["TMP"] = tempfile.gettempdir()
-    # 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(
-            [sys.executable, tmp],
-            capture_output=True,
-            text=True,
-            timeout=timeout if timeout and timeout > 0 else None,
-            env=_safe_env,
-        )
-        if proc.returncode != 0:
-            return False, (proc.stdout + "\n" + proc.stderr).strip()
-        if not os.path.exists(output_path):
-            return False, "output file was not created"
-        return True, ""
-    except subprocess.TimeoutExpired:
-        return False, f"timeout after {timeout}s"
-    finally:
+    # Keep the runner and scratch files out of the result directory. Environment
+    # scrubbing prevents direct credential inheritance; it is not a filesystem,
+    # process, or network sandbox.
+    with tempfile.TemporaryDirectory(
+        prefix="skillopt-generated-", ignore_cleanup_errors=True
+    ) as temp_dir:
+        runner = os.path.join(temp_dir, "runner.py")
+        with open(runner, "w", encoding="utf-8") as f:
+            f.write(script)
+        safe_env = generated_code_env(output_dir, temp_dir)
         try:
-            os.unlink(tmp)
-        except OSError:
-            pass
+            proc = subprocess.run(
+                [sys.executable, runner],
+                capture_output=True,
+                text=True,
+                timeout=timeout if timeout and timeout > 0 else None,
+                env=safe_env,
+            )
+            if proc.returncode != 0:
+                return False, (proc.stdout + "\n" + proc.stderr).strip()
+            if not os.path.exists(output_path):
+                return False, "output file was not created"
+            return True, ""
+        except subprocess.TimeoutExpired:
+            return False, f"timeout after {timeout}s"

skillopt/envs/spreadsheetbench/prompts/react_system.md

@@ -2,7 +2,8 @@ You are an expert spreadsheet manipulation agent.
 
 {critical_rules}{skill_section}## Tools
 You have two tools:
-- `bash` -- execute any shell command and receive its output.
+- `bash` -- run a Python command whose executable is `python` or `python3`;
+  arbitrary shell commands are blocked.
 - `write_file` -- write content to a file (path, content). Use this for solution.py.
 
 ## Protocol

skillopt/envs/spreadsheetbench/react_agent.py

@@ -12,9 +12,11 @@
 import shlex
 import subprocess
 import sys
+import tempfile
 
 from skillopt.model import chat_target_messages
 from skillopt.prompts import load_prompt
+from skillopt.envs.spreadsheetbench.executor import generated_code_env
 
 # ── Tool schemas ─────────────────────────────────────────────────────────────
 
@@ -23,13 +25,13 @@
     "function": {
         "name": "bash",
         "description": (
-            "Execute a bash command and receive stdout+stderr (truncated to 4000 chars). "
-            "Use Python to read / write Excel files."
+            "Run a Python command (python/python3 only) and receive stdout+stderr "
+            "(truncated to 4000 chars)."
         ),
         "parameters": {
             "type": "object",
             "properties": {
-                "cmd": {"type": "string", "description": "Bash command to execute."}
+                "cmd": {"type": "string", "description": "Python command to execute."}
             },
             "required": ["cmd"],
         },
@@ -40,13 +42,13 @@
     "type": "function",
     "name": "bash",
     "description": (
-        "Execute a bash command and receive stdout+stderr (truncated to 4000 chars). "
-        "Use Python to read / write Excel files."
+        "Run a Python command (python/python3 only) and receive stdout+stderr "
+        "(truncated to 4000 chars)."
     ),
     "parameters": {
         "type": "object",
         "properties": {
-            "cmd": {"type": "string", "description": "Bash command to execute."}
+            "cmd": {"type": "string", "description": "Python command to execute."}
         },
         "required": ["cmd"],
     },
@@ -279,14 +281,18 @@ def _run_bash(cmd: str, work_dir: str, timeout: int = 60) -> str:
                 "use Python to manipulate spreadsheets]"
             )
         parts[0] = sys.executable
-        proc = subprocess.run(
-            parts,
-            shell=False,
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-            cwd=work_dir,
-        )
+        with tempfile.TemporaryDirectory(
+            prefix="skillopt-generated-", ignore_cleanup_errors=True
+        ) as temp_dir:
+            proc = subprocess.run(
+                parts,
+                shell=False,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                cwd=work_dir,
+                env=generated_code_env(work_dir, temp_dir),
+            )
         out = (proc.stdout + proc.stderr).strip()
     except subprocess.TimeoutExpired:
         return f"[timeout after {timeout}s]"