Merge pull request #281 from scorp323/oracle/batch-b-safe-hardening-20260703

fix: harden tool governance and subprocess safety

tests/tools/test_base_tool_dependencies.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from tools.base_tool import BaseTool, DependencyError, ToolResult
+
+
+class DummyTool(BaseTool):
+    def execute(self, inputs: dict) -> ToolResult:
+        return ToolResult(success=True)
+
+
+class BinaryDependencyTests(unittest.TestCase):
+    def test_binary_dependency_prefix_is_checked_like_cmd(self) -> None:
+        tool = DummyTool()
+        tool.dependencies = ["binary:definitely-not-installed-openmontage-test"]
+        tool.install_instructions = "install it"
+
+        with patch("tools.base_tool.shutil.which", return_value=None):
+            with self.assertRaises(DependencyError):
+                tool.check_dependencies()
+
+    def test_binary_dependency_prefix_accepts_available_command(self) -> None:
+        tool = DummyTool()
+        tool.dependencies = ["binary:ffmpeg"]
+        tool.install_instructions = "install ffmpeg"
+
+        with patch("tools.base_tool.shutil.which", return_value="/usr/bin/ffmpeg"):
+            tool.check_dependencies()
+    def test_run_command_error_preserves_called_process_error_type(self) -> None:
+        tool = DummyTool()
+
+        with self.assertRaises(subprocess.CalledProcessError) as ctx:
+            tool.run_command([
+                sys.executable,
+                "-c",
+                "import sys; print('specific stderr', file=sys.stderr); sys.exit(7)",
+            ])
+
+        self.assertEqual(ctx.exception.returncode, 7)
+        self.assertIn("specific stderr", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/tools/test_cost_tracker_governance.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import json
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from lib.config_model import BudgetMode
+from tools.cost_tracker import CostTracker
+
+
+class CostTrackerGovernanceTests(unittest.TestCase):
+    def test_warn_mode_marks_over_budget_reservation(self) -> None:
+        with self.subTest("warning is recorded and persisted"):
+            import tempfile
+            from pathlib import Path
+
+            with tempfile.TemporaryDirectory() as temp_dir:
+                log_path = Path(temp_dir) / "cost_log.json"
+                tracker = CostTracker(
+                    budget_total_usd=1.0,
+                    reserve_pct=0.0,
+                    single_action_approval_usd=99.0,
+                    require_approval_for_new_paid_tool=False,
+                    mode=BudgetMode.WARN,
+                    cost_log_path=log_path,
+                )
+                entry_id = tracker.estimate("paid_video", "generate", 2.0)
+
+                tracker.reserve(entry_id)
+
+                entry = tracker.entries[0]
+                self.assertEqual(entry["status"], "reserved")
+                self.assertEqual(entry["reserved_usd"], 2.0)
+                self.assertTrue(entry["budget_warning"])
+                self.assertIn("exceeds usable budget", entry["budget_warning_message"])
+                persisted = json.loads(log_path.read_text())
+                self.assertTrue(persisted["entries"][0]["budget_warning"])
+
+    def test_approved_tools_persist_across_tracker_restarts(self) -> None:
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            log_path = Path(temp_dir) / "cost_log.json"
+            tracker = CostTracker(cost_log_path=log_path)
+            tracker.approve_tool("paid_video")
+
+            restarted = CostTracker(cost_log_path=log_path)
+
+            entry_id = restarted.estimate("paid_video", "generate", 0.01)
+            restarted.reserve(entry_id)
+            self.assertEqual(restarted.entries[-1]["status"], "reserved")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/tools/test_scene_detect_lavfi_escape.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from tools.analysis.scene_detect import SceneDetect
+
+
+class SceneDetectEscapingTests(unittest.TestCase):
+    def test_lavfi_movie_path_escapes_filtergraph_metacharacters(self) -> None:
+        raw = "/tmp/clip-name,with[bad];chars:01.mov"
+
+        escaped = SceneDetect._escape_lavfi_movie_path(raw)
+
+        self.assertIn("\\,", escaped)
+        self.assertIn("\\[", escaped)
+        self.assertIn("\\]", escaped)
+        self.assertIn("\\;", escaped)
+        self.assertIn("\\:", escaped)
+        self.assertNotIn("clip-name,with[bad];chars:01", escaped)
+
+    def test_lavfi_movie_path_rejects_single_quote_fail_closed(self) -> None:
+        with self.assertRaisesRegex(ValueError, "single quotes"):
+            SceneDetect._escape_lavfi_movie_path("/tmp/clip'name.mov")
+
+
+if __name__ == "__main__":
+    unittest.main()