stage-b: secure writer module updates (tests/task_plane/workers/test_codex_worker_security.py)

2026-07-25 00:02:39 +08:00
parent 063949afe5
commit 8eef5961bc
1 changed files with 70 additions and 0 deletions
@@ -0,0 +1,70 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = os.getcwd()
+import sys
+sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
+
+from app.task_plane.workers.command_policy import CommandPolicy
+from app.task_plane.workers.runner import CommandDisallowedError, SafeCommandRunner, CommandPolicyError
+from app.task_plane.workers.workspace import TaskWorkspace
+
+
+class CodexWorkerRunnerSecurityTests(unittest.TestCase):
+    def test_runner_only_allows_fixed_command_set(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-fixed", claim_id="claim-fixed") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("shell", oid="0" * 40)
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("bash", oid="0" * 40, paths=["tests/task_plane/public_interface"])
+
+    def test_runner_rejects_shell_like_and_eval_inputs(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-shell", claim_id="claim-shell") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                CommandPolicy(workspace.layout.repo_root, workspace.layout.result_root).build(
+                    "pytest",
+                    oid="0" * 40,
+                    paths=["tests/task_plane/public_interface"],
+                )
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface;touch /tmp/x"])
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "eval"])
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "bash"])
+
+    def test_runner_rejects_revision_expression_and_symlink_paths(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-revision", claim_id="claim-revision") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                policy_root = workspace.layout.repo_root
+                (policy_root / "tests" / "task_plane" / "public_interface").mkdir(parents=True, exist_ok=True)
+
+                with self.assertRaises(ValueError):
+                    runner.run("git-show", oid="HEAD~1", paths=["active-runtime/app/task_plane/worker.py"])
+
+                link_target = policy_root / "tests" / "task_plane" / "public_interface" / "link"
+                link_target.symlink_to(Path("/etc/passwd"), target_is_directory=False)
+                with self.assertRaises(CommandPolicyError):
+                    runner.run("git-show", oid="0" * 40, paths=["tests/task_plane/public_interface/link"])
+
+    def test_runner_rejects_network_tokens_and_output_escape(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-net", claim_id="claim-net") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("git-show", oid="0" * 40, paths=["https://bad.example/attack"])
+
+                with self.assertRaises(CommandPolicyError):
+                    runner.run(
+                        "acceptance",
+                        oid="0" * 40,
+                        paths=["active-runtime"],
+                        output=workspace.layout.home.parent / "escape.json",
+                    )