diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 0c9adcc..c6098e7 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -78,7 +78,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: ToolHunter_ - _Depends: 3.1, 2.2_ -- [ ] 4. Stand up the isolated sandbox runner +- [x] 4. Stand up the isolated sandbox runner - [x] 4.1 (P) Probe whether the sandbox runtime is usable - A short `docker info` style probe returns available or unavailable without throwing into the scan. - Observable completion: the fake probe can be forced off; the real probe is skippable in unit tests. diff --git a/src/openultrasast/regress/__init__.py b/src/openultrasast/regress/__init__.py new file mode 100644 index 0000000..754702a --- /dev/null +++ b/src/openultrasast/regress/__init__.py @@ -0,0 +1,13 @@ +"""Isolated regression snippets and verdicts.""" + +from .candidate import INCONCLUSIVE, SAFETY_REJECTED, RegressionRunner, RegressionVerdict +from .safety import UnsafeSnippetError, check_snippet_safety + +__all__ = [ + "INCONCLUSIVE", + "SAFETY_REJECTED", + "RegressionRunner", + "RegressionVerdict", + "UnsafeSnippetError", + "check_snippet_safety", +] diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py new file mode 100644 index 0000000..fa362cf --- /dev/null +++ b/src/openultrasast/regress/candidate.py @@ -0,0 +1,41 @@ +"""Safety-gated regression execution (task 4.3).""" + +from __future__ import annotations + +from dataclasses import dataclass + +from ..sandbox import SandboxJob, SandboxResult, SandboxRunner +from .safety import UnsafeSnippetError, check_snippet_safety + +INCONCLUSIVE = "inconclusive" +SAFETY_REJECTED = "safety_rejected" +TRIGGERABLE = "triggerable" +NOT_TRIGGERABLE = "not_triggerable" + + +@dataclass(frozen=True) +class RegressionVerdict: + verdict: str + reason: str = "" + + +class RegressionRunner: + """Run a snippet in the sandbox only after the structural safety check passes.""" + + def __init__(self, sandbox: SandboxRunner) -> None: + self._sandbox = sandbox + + def run_snippet(self, snippet: str, job: SandboxJob) -> RegressionVerdict: + try: + check_snippet_safety(snippet) + except UnsafeSnippetError: + return RegressionVerdict(verdict=INCONCLUSIVE, reason=SAFETY_REJECTED) + return _verdict_from_result(self._sandbox.run(job)) + + +def _verdict_from_result(result: SandboxResult) -> RegressionVerdict: + if result.timed_out: + return RegressionVerdict(verdict=INCONCLUSIVE, reason="timeout") + if result.exit_code != 0: + return RegressionVerdict(verdict=TRIGGERABLE, reason="nonzero_exit") + return RegressionVerdict(verdict=NOT_TRIGGERABLE, reason="exit_zero") diff --git a/src/openultrasast/regress/safety.py b/src/openultrasast/regress/safety.py new file mode 100644 index 0000000..d8dfb26 --- /dev/null +++ b/src/openultrasast/regress/safety.py @@ -0,0 +1,81 @@ +"""Structural deny-list for untrusted regression snippets. + +Snippets are data for `/scratch` inside the sandbox. Mentions of sockets, curl, +host networking, or writes under the source mount (`/workspace`) are rejected +before any `SandboxRunner.run` call. +""" + +from __future__ import annotations + +import re + +from ..sandbox.runner import WORKSPACE_MOUNT + +_WORKSPACE = re.escape(WORKSPACE_MOUNT) + + +class UnsafeSnippetError(ValueError): + """Raised when a snippet fails the structural sandbox safety check.""" + + +_HOST_NETWORK = re.compile( + r"--net(?:work)?(?:\s*=\s*|\s+|['\"]\s*,\s*['\"])host\b", + re.IGNORECASE, +) +_SOCKET_WORD = re.compile(r"\bsocket\b", re.IGNORECASE) +_CURL_WORD = re.compile(r"\bcurl\b", re.IGNORECASE) +_OPEN_WORKSPACE_WRITE = re.compile( + rf"open\s*\(\s*[fFrRbBuU]*['\"]{_WORKSPACE}[^'\"]*['\"]\s*,\s*(?:mode\s*=\s*)?[fFrRbBuU]*['\"][^'\"]*[wax+]", + re.IGNORECASE | re.DOTALL, +) +_PATH_WORKSPACE_WRITE = re.compile( + rf"Path\s*\(\s*['\"]{_WORKSPACE}[^'\"]*['\"]\s*\)\s*\.\s*" + r"(?:write_text|write_bytes|touch|mkdir|unlink|rmdir|replace|chmod|rename|open)\b", + re.IGNORECASE | re.DOTALL, +) +_FS_WORKSPACE_WRITE = re.compile( + r"(?:writeFile(?:Sync)?|appendFile(?:Sync)?|mkdir(?:Sync)?|rm(?:dir)?(?:Sync)?|" + rf"unlink(?:Sync)?|rename(?:Sync)?)\s*\(\s*['\"]{_WORKSPACE}", + re.IGNORECASE, +) +_FOPEN_WORKSPACE_WRITE = re.compile( + rf"fopen\s*\(\s*['\"]{_WORKSPACE}[^'\"]*['\"]\s*,\s*['\"][^'\"]*[wax+]", + re.IGNORECASE | re.DOTALL, +) +_SHUTIL_OS_WORKSPACE_WRITE = re.compile( + r"(?:shutil\.(?:copy(?:2)?|copyfile|copytree|move|rmtree)|" + r"os\.(?:remove|unlink|replace|rename|mkdir|makedirs|rmdir|chmod))" + rf"\s*\([^;]{{0,240}}{_WORKSPACE}", + re.IGNORECASE | re.DOTALL, +) +_REDIRECT_WORKSPACE = re.compile(rf"(?:>>|>)\s*['\"]?{_WORKSPACE}\b", re.IGNORECASE) + +_WORKSPACE_WRITES = ( + _OPEN_WORKSPACE_WRITE, + _PATH_WORKSPACE_WRITE, + _FS_WORKSPACE_WRITE, + _FOPEN_WORKSPACE_WRITE, + _SHUTIL_OS_WORKSPACE_WRITE, + _REDIRECT_WORKSPACE, +) + + +def check_snippet_safety(snippet: str) -> None: + """Return None when `snippet` is structurally safe; raise otherwise.""" + lowered = snippet.lower() + if "docker.sock" in lowered: + raise UnsafeSnippetError("snippet mentions the Docker socket") + if _SOCKET_WORD.search(snippet) is not None: + raise UnsafeSnippetError("snippet mentions socket") + if _CURL_WORD.search(snippet) is not None: + raise UnsafeSnippetError("snippet mentions curl") + if _HOST_NETWORK.search(snippet) is not None: + raise UnsafeSnippetError("snippet requests host network") + if _writes_under_workspace(snippet): + raise UnsafeSnippetError(f"snippet writes under {WORKSPACE_MOUNT}") + + +def _writes_under_workspace(snippet: str) -> bool: + if WORKSPACE_MOUNT not in snippet.lower(): + return False + return any(pattern.search(snippet) is not None for pattern in _WORKSPACE_WRITES) diff --git a/tests/test_regress_safety.py b/tests/test_regress_safety.py new file mode 100644 index 0000000..c8360e3 --- /dev/null +++ b/tests/test_regress_safety.py @@ -0,0 +1,86 @@ +"""Reject unsafe regression snippets before the sandbox runner sees them (task 4.3).""" + +from pathlib import Path + +import pytest + +from openultrasast.regress import RegressionRunner, UnsafeSnippetError, check_snippet_safety +from openultrasast.sandbox import FakeSandboxRunner, SandboxJob + +BENIGN_ASSERT = "assert 1 + 1 == 2\n" + + +def _job(tmp_path: Path, snippet: str) -> SandboxJob: + repo = tmp_path / "repo" + repo.mkdir() + return SandboxJob( + image="python:3.12-alpine", + command=("python", "/scratch/case.py"), + repo_root=repo, + scratch_files={"case.py": snippet}, + timeout_seconds=30, + memory_mb=256, + pids_limit=128, + ) + + +def test_docker_sock_rejected_does_not_create_runner_job(tmp_path: Path) -> None: + snippet = 'client = docker.DockerClient(base_url="unix://var/run/docker.sock")\n' + sandbox = FakeSandboxRunner() + runner = RegressionRunner(sandbox) + + with pytest.raises(UnsafeSnippetError, match="Docker socket"): + check_snippet_safety(snippet) + verdict = runner.run_snippet(snippet, _job(tmp_path, snippet)) + + assert verdict.verdict == "inconclusive" + assert verdict.reason == "safety_rejected" + assert sandbox.jobs == [] + + +def test_host_network_rejected_does_not_create_runner_job(tmp_path: Path) -> None: + snippet = "subprocess.run(['docker', 'run', '--network', 'host', 'alpine'])\n" + sandbox = FakeSandboxRunner() + runner = RegressionRunner(sandbox) + + with pytest.raises(UnsafeSnippetError, match="host network"): + check_snippet_safety(snippet) + verdict = runner.run_snippet(snippet, _job(tmp_path, snippet)) + + assert verdict.verdict == "inconclusive" + assert verdict.reason == "safety_rejected" + assert sandbox.jobs == [] + + +def test_write_under_workspace_rejected_does_not_create_runner_job(tmp_path: Path) -> None: + snippet = 'open("/workspace/pwned.py", "w").write("owned")\n' + sandbox = FakeSandboxRunner() + runner = RegressionRunner(sandbox) + + with pytest.raises(UnsafeSnippetError, match="/workspace"): + check_snippet_safety(snippet) + verdict = runner.run_snippet(snippet, _job(tmp_path, snippet)) + + assert verdict.verdict == "inconclusive" + assert verdict.reason == "safety_rejected" + assert sandbox.jobs == [] + + +def test_benign_python_assert_snippet_is_allowed(tmp_path: Path) -> None: + assert check_snippet_safety(BENIGN_ASSERT) is None + sandbox = FakeSandboxRunner() + job = _job(tmp_path, BENIGN_ASSERT) + RegressionRunner(sandbox).run_snippet(BENIGN_ASSERT, job) + assert sandbox.jobs == [job] + + +@pytest.mark.parametrize( + "snippet", + [ + "import socket\nsocket.create_connection(('127.0.0.1', 80))\n", + "curl https://example.invalid/pwn\n", + ], +) +def test_socket_and_curl_mentions_are_rejected(snippet: str) -> None: + with pytest.raises(UnsafeSnippetError): + check_snippet_safety(snippet)