diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 467e052..bd74f37 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -36,3 +36,7 @@ **Vulnerability:** Argument Injection via relative paths starting with a hyphen in command-line utilities. **Learning:** Even when `ffmpeg` inputs are protected by `-i`, the output paths, as well as arguments to other utilities like `brctl` and `SetFile`, can be maliciously crafted to start with `-` and be interpreted as options if relative paths are used. **Prevention:** Resolve file paths before passing them to `subprocess.run` when a tool does not support an explicit input flag or `--` delimiter. Absolute paths use a root, drive, or UNC prefix rather than a leading hyphen, so they cannot be parsed as command-line options. +## 2024-06-25 - [Sentinel: Uncontrolled Resource Consumption in Subprocesses] +**Vulnerability:** Uncontrolled Resource Consumption (CWE-400) / DoS via hung external processes. +**Learning:** When using `subprocess.run` to execute external binaries (like `ffmpeg`, `ffprobe`, or `brctl`), the application can hang indefinitely if the external process stalls. This leads to resource exhaustion (e.g., worker threads/processes blocked) and potential Denial of Service. +**Prevention:** Always configure an explicit `timeout` parameter and handle `subprocess.TimeoutExpired` exceptions for all `subprocess.run` calls. Tailor the timeout duration to the specific task (e.g., 60s for `ffprobe`, 3600s for `ffmpeg` transcoding). diff --git a/media_shrinker.py b/media_shrinker.py index 7ee2dfb..a9b5d8e 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -465,7 +465,10 @@ def probe_media( "-i", str(source_path), ] - completed = subprocess.run(command, check=False, capture_output=True, text=True) + try: + completed = subprocess.run(command, check=False, capture_output=True, text=True, timeout=60) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffprobe timed out for {source_path}") from exc if completed.returncode != 0: raise MediaShrinkerError( f"ffprobe failed for {source_path}: {completed.stderr.strip()}" @@ -518,17 +521,21 @@ def detect_silence_intervals( ) -> list[SilenceInterval]: """Run ffmpeg silencedetect and return paired silence intervals.""" - completed = subprocess.run( - build_silencedetect_command( - source_path, - ffmpeg_path=ffmpeg_path, - silence_noise=silence_noise, - silence_min_duration_seconds=silence_min_duration_seconds, - ), - check=False, - capture_output=True, - text=True, - ) + try: + completed = subprocess.run( + build_silencedetect_command( + source_path, + ffmpeg_path=ffmpeg_path, + silence_noise=silence_noise, + silence_min_duration_seconds=silence_min_duration_seconds, + ), + check=False, + capture_output=True, + text=True, + timeout=600, + ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"silencedetect timed out for {source_path}") from exc if completed.returncode != 0: raise MediaShrinkerError( f"silencedetect failed for {source_path}: {completed.stderr.strip()}" @@ -643,12 +650,16 @@ def download_from_icloud(source_path: Path, *, brctl_path: str = "brctl") -> Non raise MediaShrinkerError( f"iCloud download requested but '{brctl_path}' was not found" ) - completed = subprocess.run( - build_icloud_download_command(source_path, brctl_path=brctl_path), - check=False, - capture_output=True, - text=True, - ) + try: + completed = subprocess.run( + build_icloud_download_command(source_path, brctl_path=brctl_path), + check=False, + capture_output=True, + text=True, + timeout=600, + ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"iCloud download timed out for {source_path}") from exc if completed.returncode != 0: raise MediaShrinkerError( f"iCloud download failed for {source_path}: {completed.stderr.strip()}" @@ -1586,8 +1597,10 @@ def _execute_plan( ) try: completed = subprocess.run( - command, check=False, capture_output=True, text=True + command, check=False, capture_output=True, text=True, timeout=3600 ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffmpeg timed out for {source}") from exc except FileNotFoundError as exc: raise MediaShrinkerError(f"ffmpeg not found: {ffmpeg_path}") from exc @@ -1654,12 +1667,16 @@ def _copy_macos_creation_time( creation_date = datetime.fromtimestamp(float(birthtime)).strftime( "%m/%d/%Y %H:%M:%S" ) - subprocess.run( - [setfile_path, "-d", creation_date, str(dest.resolve())], - check=False, - capture_output=True, - text=True, - ) + try: + subprocess.run( + [setfile_path, "-d", creation_date, str(dest.resolve())], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + except subprocess.TimeoutExpired: + pass def _format_result(root: Path, result: ConversionResult) -> str: diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 7f5fc24..1f3148b 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -1684,3 +1684,59 @@ def test_attribute_copy_helpers_ignore_platform_errors(self) -> None: if __name__ == "__main__": unittest.main() + +class TestSubprocessTimeouts(unittest.TestCase): + @patch("media_shrinker.subprocess.run") + def test_ffprobe_timeout(self, mock_run) -> None: + from media_shrinker import probe_media, MediaShrinkerError + import subprocess + mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffprobe", timeout=60) + with self.assertRaisesRegex(MediaShrinkerError, "ffprobe timed out for"): + probe_media(Path("dummy.wav")) + + @patch("media_shrinker.subprocess.run") + def test_silencedetect_timeout(self, mock_run) -> None: + from media_shrinker import detect_silence_intervals, MediaShrinkerError + import subprocess + mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=3600) + with self.assertRaisesRegex(MediaShrinkerError, "silencedetect timed out for"): + detect_silence_intervals(Path("dummy.wav")) + + @patch("media_shrinker.subprocess.run") + def test_icloud_download_timeout(self, mock_run) -> None: + from media_shrinker import download_from_icloud, MediaShrinkerError + import subprocess + mock_run.side_effect = subprocess.TimeoutExpired(cmd="brctl", timeout=3600) + with self.assertRaisesRegex(MediaShrinkerError, "iCloud download timed out for"): + download_from_icloud(Path("dummy.wav"), brctl_path="ls") + + @patch("media_shrinker.subprocess.run") + def test_ffmpeg_timeout(self, mock_run) -> None: + from media_shrinker import _execute_plan, ConversionPlan, MediaShrinkerError + import subprocess + mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=3600) + plan = ConversionPlan( + strategy="flac-lossless", + input_path=Path("dummy.wav"), + output_path=Path("dummy.flac"), + ffmpeg_args=["-i", "dummy.wav", "-c:a", "flac", "dummy.flac"] + ) + with self.assertRaisesRegex(MediaShrinkerError, "ffmpeg timed out for"): + _execute_plan( + plan, ffmpeg_path="ffmpeg", source=Path("dummy.wav"), final_output=Path("dummy.flac"), + protected_sources=set(), overwrite=True + ) + + @patch("media_shrinker.subprocess.run") + def test_setfile_timeout(self, mock_run) -> None: + from media_shrinker import _copy_macos_creation_time + import subprocess + import os + from unittest.mock import Mock + mock_run.side_effect = subprocess.TimeoutExpired(cmd="SetFile", timeout=60) + + stat_mock = Mock(spec=os.stat_result) + stat_mock.st_birthtime = 1000.0 + + _copy_macos_creation_time(stat_mock, Path("dummy.wav"), "SetFile") + # Should catch and ignore the timeout, so no exception is raised