From 6580987931938c5d3e8bd336a2dacbbb572882e1 Mon Sep 17 00:00:00 2001 From: Morpheus Date: Fri, 3 Jul 2026 20:07:00 +0800 Subject: [PATCH 1/2] fix: harden tool governance and subprocess safety - persist budget warnings and approved paid-tool decisions - support binary dependency declarations - include stderr/stdout details for failed subprocesses - escape lavfi movie paths used by ffmpeg scene detection Verification: - python3 tests/tools/test_cost_tracker_governance.py - python3 tests/tools/test_scene_detect_lavfi_escape.py - python3 tests/tools/test_base_tool_dependencies.py - python3 -m py_compile ... --- tests/tools/test_base_tool_dependencies.py | 48 +++++++++++++++ tests/tools/test_cost_tracker_governance.py | 60 +++++++++++++++++++ tests/tools/test_scene_detect_lavfi_escape.py | 29 +++++++++ tools/analysis/scene_detect.py | 15 ++++- tools/base_tool.py | 41 ++++++++----- tools/cost_tracker.py | 15 +++-- 6 files changed, 187 insertions(+), 21 deletions(-) create mode 100644 tests/tools/test_base_tool_dependencies.py create mode 100644 tests/tools/test_cost_tracker_governance.py create mode 100644 tests/tools/test_scene_detect_lavfi_escape.py diff --git a/tests/tools/test_base_tool_dependencies.py b/tests/tools/test_base_tool_dependencies.py new file mode 100644 index 00000000..e618bd59 --- /dev/null +++ b/tests/tools/test_base_tool_dependencies.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +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_includes_stderr(self) -> None: + tool = DummyTool() + + with self.assertRaisesRegex(RuntimeError, "specific stderr"): + tool.run_command([ + sys.executable, + "-c", + "import sys; print('specific stderr', file=sys.stderr); sys.exit(7)", + ]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/tools/test_cost_tracker_governance.py b/tests/tools/test_cost_tracker_governance.py new file mode 100644 index 00000000..e97e4783 --- /dev/null +++ b/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() diff --git a/tests/tools/test_scene_detect_lavfi_escape.py b/tests/tools/test_scene_detect_lavfi_escape.py new file mode 100644 index 00000000..b7ee600d --- /dev/null +++ b/tests/tools/test_scene_detect_lavfi_escape.py @@ -0,0 +1,29 @@ +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.assertIn("\\:", escaped) + self.assertNotIn("clip'name,with[bad];chars:01", escaped) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/analysis/scene_detect.py b/tools/analysis/scene_detect.py index 71062a4f..f0d278e1 100644 --- a/tools/analysis/scene_detect.py +++ b/tools/analysis/scene_detect.py @@ -164,11 +164,24 @@ class SceneDetect(BaseTool): return scenes + @staticmethod + def _escape_lavfi_movie_path(path: str) -> str: + """Escape a path for FFmpeg lavfi movie=... without allowing filter injection.""" + normalized = path.replace("\\", "/") + escaped = [] + for char in normalized: + if char in "\\':,[];": + escaped.append("\\" + char) + else: + escaped.append(char) + return "".join(escaped) + def _detect_ffmpeg(self, inputs: dict[str, Any]) -> list[dict]: """Fallback: use FFmpeg scene change filter.""" input_path = str(inputs["input_path"]) threshold = inputs.get("threshold", 0.3) min_scene_len = inputs.get("min_scene_length_seconds", 1.0) + escaped_input = self._escape_lavfi_movie_path(input_path) cmd = [ "ffprobe", @@ -176,7 +189,7 @@ class SceneDetect(BaseTool): "-show_entries", "frame=pts_time", "-of", "json", "-f", "lavfi", - f"movie='{input_path.replace(chr(92), '/').replace(':', chr(92)+':')}',select='gt(scene,{threshold})'", + f"movie='{escaped_input}',select='gt(scene,{threshold})'", ] try: diff --git a/tools/base_tool.py b/tools/base_tool.py index 50e6d249..50e3b850 100644 --- a/tools/base_tool.py +++ b/tools/base_tool.py @@ -209,8 +209,9 @@ class BaseTool(ABC): def check_dependencies(self) -> None: """Verify all dependencies are installed. Raises DependencyError if not.""" for dep in self.dependencies: - if dep.startswith("cmd:"): - cmd_name = dep[4:] + if dep.startswith(("cmd:", "binary:")): + prefix = "cmd:" if dep.startswith("cmd:") else "binary:" + cmd_name = dep[len(prefix):] if shutil.which(cmd_name) is None: raise DependencyError( f"Command {cmd_name!r} not found. {self.install_instructions}" @@ -329,20 +330,28 @@ class BaseTool(ABC): exe = shutil.which(resolved_cmd[0]) if exe: resolved_cmd[0] = exe - return subprocess.run( - resolved_cmd, - capture_output=True, - text=True, - # Force UTF-8 decoding. The default uses the OS locale (cp1252 on - # Windows), which raises UnicodeDecodeError on a subprocess that - # emits Unicode/emoji (e.g. Remotion's progress output), killing the - # reader thread and potentially swallowing the real error text. - encoding="utf-8", - errors="replace", - timeout=timeout, - cwd=cwd, - check=True, - ) + try: + return subprocess.run( + resolved_cmd, + capture_output=True, + text=True, + # Force UTF-8 decoding. The default uses the OS locale (cp1252 on + # Windows), which raises UnicodeDecodeError on a subprocess that + # emits Unicode/emoji (e.g. Remotion's progress output), killing the + # reader thread and potentially swallowing the real error text. + encoding="utf-8", + errors="replace", + timeout=timeout, + cwd=cwd, + check=True, + ) + except subprocess.CalledProcessError as exc: + stderr = (exc.stderr or "").strip() + stdout = (exc.stdout or "").strip() + detail = stderr or stdout or str(exc) + raise RuntimeError( + f"Command failed with exit code {exc.returncode}: {' '.join(resolved_cmd)}\n{detail}" + ) from exc class DependencyError(Exception): diff --git a/tools/cost_tracker.py b/tools/cost_tracker.py index ee324efb..8c363d7d 100644 --- a/tools/cost_tracker.py +++ b/tools/cost_tracker.py @@ -141,11 +141,15 @@ class CostTracker: # Check budget if estimated > self.usable_budget_usd: + message = ( + f"Reservation of ${estimated:.2f} exceeds usable budget " + f"${self.usable_budget_usd:.2f}" + ) if self.mode == BudgetMode.CAP: - raise BudgetExceededError( - f"Reservation of ${estimated:.2f} exceeds usable budget " - f"${self.usable_budget_usd:.2f}" - ) + raise BudgetExceededError(message) + if self.mode == BudgetMode.WARN: + entry["budget_warning"] = True + entry["budget_warning_message"] = message entry["status"] = EntryStatus.RESERVED.value entry["reserved_usd"] = estimated @@ -155,6 +159,7 @@ class CostTracker: def approve_tool(self, tool: str) -> None: """Mark a tool as approved for paid operations.""" self._approved_tools.add(tool) + self._save() def reconcile(self, entry_id: str, actual_usd: float, success: bool = True) -> None: """Reconcile actual spend after tool execution.""" @@ -487,6 +492,7 @@ class CostTracker: "budget_total_usd": self.budget_total_usd, "budget_reserved_usd": round(self.budget_reserved_usd, 4), "budget_spent_usd": round(self.budget_spent_usd, 4), + "approved_tools": sorted(self._approved_tools), "entries": self.entries, } self.cost_log_path.parent.mkdir(parents=True, exist_ok=True) @@ -498,6 +504,7 @@ class CostTracker: data = json.load(f) self.entries = data.get("entries", []) self.budget_total_usd = data.get("budget_total_usd", self.budget_total_usd) + self._approved_tools = set(data.get("approved_tools", [])) # ---- Helpers ---- From 14ebc56123fdecaac799917075f91b4ed03bc40f Mon Sep 17 00:00:00 2001 From: Morpheus Date: Fri, 3 Jul 2026 22:00:38 +0800 Subject: [PATCH 2/2] fix: preserve subprocess error type in tool runner - raise a CalledProcessError subclass that surfaces stderr/stdout detail - keep existing callers that catch subprocess.CalledProcessError working - reject lavfi movie paths containing single quotes fail-closed - add regression coverage for both review findings Verification: - python3 -m unittest tests.tools.test_base_tool_dependencies tests.tools.test_scene_detect_lavfi_escape tests.tools.test_cost_tracker_governance - python3 -m py_compile tools/base_tool.py tools/analysis/scene_detect.py tools/cost_tracker.py tests/tools/test_base_tool_dependencies.py tests/tools/test_scene_detect_lavfi_escape.py tests/tools/test_cost_tracker_governance.py --- tests/tools/test_base_tool_dependencies.py | 8 +++-- tests/tools/test_scene_detect_lavfi_escape.py | 9 ++++-- tools/analysis/scene_detect.py | 4 ++- tools/base_tool.py | 30 +++++++++++++++++-- 4 files changed, 43 insertions(+), 8 deletions(-) diff --git a/tests/tools/test_base_tool_dependencies.py b/tests/tools/test_base_tool_dependencies.py index e618bd59..79505a2a 100644 --- a/tests/tools/test_base_tool_dependencies.py +++ b/tests/tools/test_base_tool_dependencies.py @@ -1,5 +1,6 @@ from __future__ import annotations +import subprocess import sys import unittest from pathlib import Path @@ -33,16 +34,19 @@ class BinaryDependencyTests(unittest.TestCase): with patch("tools.base_tool.shutil.which", return_value="/usr/bin/ffmpeg"): tool.check_dependencies() - def test_run_command_error_includes_stderr(self) -> None: + def test_run_command_error_preserves_called_process_error_type(self) -> None: tool = DummyTool() - with self.assertRaisesRegex(RuntimeError, "specific stderr"): + 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() diff --git a/tests/tools/test_scene_detect_lavfi_escape.py b/tests/tools/test_scene_detect_lavfi_escape.py index b7ee600d..717b1611 100644 --- a/tests/tools/test_scene_detect_lavfi_escape.py +++ b/tests/tools/test_scene_detect_lavfi_escape.py @@ -12,17 +12,20 @@ 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" + 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.assertIn("\\:", escaped) - self.assertNotIn("clip'name,with[bad];chars:01", 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__": diff --git a/tools/analysis/scene_detect.py b/tools/analysis/scene_detect.py index f0d278e1..8ca2ecb5 100644 --- a/tools/analysis/scene_detect.py +++ b/tools/analysis/scene_detect.py @@ -168,9 +168,11 @@ class SceneDetect(BaseTool): def _escape_lavfi_movie_path(path: str) -> str: """Escape a path for FFmpeg lavfi movie=... without allowing filter injection.""" normalized = path.replace("\\", "/") + if "'" in normalized: + raise ValueError("FFmpeg lavfi movie paths containing single quotes are unsupported") escaped = [] for char in normalized: - if char in "\\':,[];": + if char in "\\:,[];": escaped.append("\\" + char) else: escaped.append(char) diff --git a/tools/base_tool.py b/tools/base_tool.py index 50e3b850..8e89f9a6 100644 --- a/tools/base_tool.py +++ b/tools/base_tool.py @@ -349,11 +349,37 @@ class BaseTool(ABC): stderr = (exc.stderr or "").strip() stdout = (exc.stdout or "").strip() detail = stderr or stdout or str(exc) - raise RuntimeError( - f"Command failed with exit code {exc.returncode}: {' '.join(resolved_cmd)}\n{detail}" + raise ToolCommandError( + exc.returncode, + exc.cmd, + output=exc.output, + stderr=exc.stderr, + detail=detail, ) from exc +class ToolCommandError(subprocess.CalledProcessError): + """CalledProcessError with stderr/stdout surfaced in str(error).""" + + def __init__( + self, + returncode: int, + cmd: list[str], + *, + output: Optional[str] = None, + stderr: Optional[str] = None, + detail: str = "", + ) -> None: + super().__init__(returncode, cmd, output=output, stderr=stderr) + self.detail = detail + + def __str__(self) -> str: + base = super().__str__() + if self.detail: + return f"{base}\n{self.detail}" + return base + + class DependencyError(Exception): """Raised when a tool's dependency is not satisfied.""" pass