mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-28 13:26:48 +00:00
caa3adbbce
Reimplements the useful parts of #385 cleanly. Adds two small cross-platform helpers under `openjarvis.core.utils`: - `get_python_executable()` — prefers `python3`, falls back to `python` for Windows / minimal distros that only ship the unversioned name. - `open_browser(url)` — `webbrowser.open` by default; on Windows uses `cmd /c start "" <url>` to avoid console-host edge cases. Swapped at every hardcoded `python3` / `webbrowser.open` site: `connectors/oauth.py`, `evals/scorers/livecodebench.py`, `scripts/oauth_all.py`, `scripts/install/install.sh` (adds `PY_CMD` detection block), `scripts/quickstart.sh` (adds `MINGW*|MSYS*|CYGWIN*) cmd /c start` case), and the two affected test files. Test files wrap `get_python_executable()` in `shlex.quote()` before interpolating into `shell=True` strings — Windows interpreter paths often contain spaces. Deliberately different from #385: `openjarvis.core.__init__` does NOT re-export `DEFAULT_CONFIG_DIR` (would have raised ImportError because it's in `openjarvis.core.config`, not the package `__init__`; re-exporting would also force eager import of the heavy config module at every `import openjarvis.core`). `oauth.py` keeps `from openjarvis.core.config import DEFAULT_CONFIG_DIR` alongside the new `from openjarvis.core import open_browser`. Original API surface and call-site sweep by @sanjayravit in #385 — huge thanks for the careful Windows-compatibility audit. This PR preserves your design while fixing the ImportError edge cases caught during review. Co-Authored-By: sanjayravit <sanjayravit@users.noreply.github.com> Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
"""Tests for subprocess sandbox — secure process execution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shlex
|
|
import tempfile
|
|
|
|
from openjarvis.core import get_python_executable
|
|
from openjarvis.security.subprocess_sandbox import (
|
|
build_safe_env,
|
|
kill_process_tree,
|
|
run_sandboxed,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_safe_env tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBuildSafeEnv:
|
|
def test_only_safe_vars_included(self) -> None:
|
|
env = build_safe_env()
|
|
# All keys should be from the safe set
|
|
safe_keys = {
|
|
"PATH",
|
|
"HOME",
|
|
"USER",
|
|
"LANG",
|
|
"TERM",
|
|
"SHELL",
|
|
"LC_ALL",
|
|
"LC_CTYPE",
|
|
"TMPDIR",
|
|
"TZ",
|
|
}
|
|
for key in env:
|
|
assert key in safe_keys
|
|
|
|
def test_passthrough_works(self) -> None:
|
|
os.environ["MY_CUSTOM_VAR_FOR_TEST"] = "hello"
|
|
try:
|
|
env = build_safe_env(passthrough=["MY_CUSTOM_VAR_FOR_TEST"])
|
|
assert env.get("MY_CUSTOM_VAR_FOR_TEST") == "hello"
|
|
finally:
|
|
del os.environ["MY_CUSTOM_VAR_FOR_TEST"]
|
|
|
|
def test_extra_vars_added(self) -> None:
|
|
env = build_safe_env(extra={"FOO": "bar", "BAZ": "qux"})
|
|
assert env["FOO"] == "bar"
|
|
assert env["BAZ"] == "qux"
|
|
|
|
def test_unknown_env_var_excluded(self) -> None:
|
|
os.environ["SUPER_SECRET_KEY_XYZ"] = "secret"
|
|
try:
|
|
env = build_safe_env()
|
|
assert "SUPER_SECRET_KEY_XYZ" not in env
|
|
finally:
|
|
del os.environ["SUPER_SECRET_KEY_XYZ"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_sandboxed tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunSandboxed:
|
|
def test_simple_echo(self) -> None:
|
|
result = run_sandboxed("echo hello", timeout=10.0)
|
|
assert result.returncode == 0
|
|
assert "hello" in result.stdout
|
|
assert not result.timed_out
|
|
assert not result.killed
|
|
|
|
def test_timeout_kills_process(self) -> None:
|
|
result = run_sandboxed("sleep 60", timeout=1.0)
|
|
assert result.timed_out
|
|
assert result.killed
|
|
assert result.returncode == -1
|
|
|
|
def test_working_dir(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
result = run_sandboxed("pwd", working_dir=tmpdir, timeout=10.0)
|
|
assert result.returncode == 0
|
|
assert tmpdir in result.stdout.strip()
|
|
|
|
def test_env_isolation(self) -> None:
|
|
os.environ["TEST_SECRET"] = "super_secret_value"
|
|
try:
|
|
result = run_sandboxed(
|
|
'echo "val=$TEST_SECRET"',
|
|
timeout=10.0,
|
|
)
|
|
assert result.returncode == 0
|
|
assert "super_secret_value" not in result.stdout
|
|
finally:
|
|
del os.environ["TEST_SECRET"]
|
|
|
|
def test_output_truncation(self) -> None:
|
|
# Generate output larger than max_output_bytes
|
|
result = run_sandboxed(
|
|
f"{shlex.quote(get_python_executable())} -c \"print('A' * 200)\"",
|
|
timeout=10.0,
|
|
max_output_bytes=50,
|
|
)
|
|
assert result.returncode == 0
|
|
assert len(result.stdout) <= 50
|
|
|
|
def test_non_zero_exit_code(self) -> None:
|
|
result = run_sandboxed("exit 42", timeout=10.0)
|
|
assert result.returncode == 42
|
|
assert not result.timed_out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# kill_process_tree tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestKillProcessTree:
|
|
def test_no_crash_on_nonexistent_pid(self) -> None:
|
|
# Should not raise on a PID that doesn't exist
|
|
kill_process_tree(999999999)
|