From 14ebc56123fdecaac799917075f91b4ed03bc40f Mon Sep 17 00:00:00 2001 From: Morpheus Date: Fri, 3 Jul 2026 22:00:38 +0800 Subject: [PATCH] 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