diff --git a/.gitignore b/.gitignore index 8d89812..fd9766d 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ *.py[cod] .DS_Store .coverage +venv/ diff --git a/.jules/sentinel.md b/.jules/sentinel.md index a0d3824..e3189df 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -46,3 +46,7 @@ **Vulnerability:** Argument Injection via relative paths starting with a hyphen in command-line utilities (CWE-88). **Learning:** Even when `ffmpeg` inputs are protected by `-i`, command-line utilities (like `ffprobe` and `ffmpeg` filters) can interpret user input (like a file path) starting with a hyphen (e.g., `-version.wav`) as options if passed as a relative path. **Prevention:** File paths must be converted to absolute paths using `.resolve()` before they are passed to `subprocess.run`. This prefixes the path with a root, drive, or UNC prefix rather than a leading hyphen, thereby averting the possibility of argument injection. +## 2026-07-06 - [Sentinel: Uncontrolled Resource Consumption (DoS) via Subprocess Timeouts] +**Vulnerability:** Uncontrolled Resource Consumption (CWE-400) via missing subprocess timeouts. +**Learning:** `subprocess.run` calls without explicit `timeout` arguments can cause the application to hang indefinitely if the spawned process (e.g., `ffmpeg`, `ffprobe`, `brctl`) deadlocks or takes an unreasonable amount of time due to maliciously crafted input files or underlying system issues. +**Prevention:** Always specify an explicit, appropriate `timeout` parameter for `subprocess.run` calls (e.g., 60s for probes/metadata, 3600s+ for intensive processing) and handle the resulting `subprocess.TimeoutExpired` exception to ensure the application fails securely and releases resources. diff --git a/media_shrinker.py b/media_shrinker.py index d72a8d5..b216d59 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -465,9 +465,12 @@ def probe_media( "-i", str(Path(source_path).resolve()), ] - completed = subprocess.run( - command, check=False, capture_output=True, text=True, shell=False - ) + try: + completed = subprocess.run( + command, check=False, capture_output=True, text=True, shell=False, 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()}" @@ -520,18 +523,22 @@ 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, - shell=False, - ) + 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, + shell=False, + timeout=3600, + ) + 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()}" @@ -646,13 +653,17 @@ 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, - shell=False, - ) + try: + completed = subprocess.run( + build_icloud_download_command(source_path, brctl_path=brctl_path), + check=False, + capture_output=True, + text=True, + shell=False, + timeout=3600, + ) + 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()}" @@ -1590,10 +1601,12 @@ def _execute_plan( ) try: completed = subprocess.run( - command, check=False, capture_output=True, text=True, shell=False + command, check=False, capture_output=True, text=True, shell=False, timeout=3600 ) except FileNotFoundError as exc: raise MediaShrinkerError(f"ffmpeg not found: {ffmpeg_path}") from exc + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffmpeg timed out for {source}") from exc if completed.returncode != 0: raise MediaShrinkerError( @@ -1658,13 +1671,17 @@ 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, - shell=False, - ) + try: + subprocess.run( + [setfile_path, "-d", creation_date, str(dest.resolve())], + check=False, + capture_output=True, + text=True, + shell=False, + timeout=60, + ) + except subprocess.TimeoutExpired: + pass # Metadata restoration failure is non-fatal def _format_result(root: Path, result: ConversionResult) -> str: diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 0ab2397..7fbf824 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -1705,7 +1705,6 @@ def test_attribute_copy_helpers_ignore_platform_errors(self) -> None: class FastPathTests(unittest.TestCase): def test_copy_extended_attributes_dummy(self) -> None: - import os from media_shrinker import _copy_extended_attributes # We need to hit lines 703-704 @@ -1720,7 +1719,6 @@ def test_copy_extended_attributes_dummy(self) -> None: def test_copy_macos_creation_time_dummy(self) -> None: from media_shrinker import _copy_macos_creation_time - import stat # Hit 1632 class MockStat: @@ -1747,7 +1745,6 @@ def test_format_result_dummy(self) -> None: self.assertIn("foo.txt", s) def test_copy_extended_attributes_dummy_success(self) -> None: - import os from media_shrinker import _copy_extended_attributes # Hit 703-704 @@ -1763,7 +1760,6 @@ def test_copy_extended_attributes_dummy_success(self) -> None: def test_copy_macos_creation_time_dummy_none(self) -> None: from media_shrinker import _copy_macos_creation_time - import stat # Hit 1632 class MockStat: @@ -1775,7 +1771,6 @@ class MockStat: _copy_macos_creation_time(MockStat(), dest, "/bin/echo") def test_copy_extended_attributes_dummy_set_fail(self) -> None: - import os from media_shrinker import _copy_extended_attributes # Hit 703-704 @@ -1792,7 +1787,6 @@ def test_copy_extended_attributes_dummy_set_fail(self) -> None: def test_copy_macos_creation_time_dummy_not_found(self) -> None: from media_shrinker import _copy_macos_creation_time - import stat # Hit 1632 class MockStat: @@ -1807,7 +1801,6 @@ class MockStat: def test_copy_macos_creation_time_dummy_success(self) -> None: from media_shrinker import _copy_macos_creation_time - import stat # Hit 1632 class MockStat: @@ -1816,7 +1809,7 @@ class MockStat: with tempfile.TemporaryDirectory() as tmp: dest = Path(tmp) / "dest.txt" dest.write_text("hello") - with patch("subprocess.run") as mock_run: + with patch("subprocess.run"): _copy_macos_creation_time(MockStat(), dest, "/bin/echo") def test_preserve_file_attributes_no_setfile(self) -> None: @@ -1838,7 +1831,6 @@ class MockStat: preserve_file_attributes(src, dest) def test_copy_extended_attributes_dummy_success_branch(self) -> None: - import os from media_shrinker import _copy_extended_attributes # Hit 703-704 @@ -1856,7 +1848,6 @@ def test_copy_extended_attributes_dummy_success_branch(self) -> None: def test_copy_extended_attributes_dummy_listxattr_missing(self) -> None: import builtins - import os from media_shrinker import _copy_extended_attributes # Hit early return inside _copy_extended_attributes when OS doesn't support it @@ -1877,7 +1868,6 @@ def fake_hasattr(obj, name): def test_preserve_file_attributes_chmod_error(self) -> None: from media_shrinker import preserve_file_attributes - import stat # Hit 703-704 class MockStat: @@ -1897,7 +1887,6 @@ class MockStat: def test_preserve_file_attributes_with_setfile(self) -> None: from media_shrinker import preserve_file_attributes - import stat # Hit 703-704 class MockStat: @@ -1914,3 +1903,87 @@ class MockStat: with patch("media_shrinker._copy_macos_creation_time"): with patch("media_shrinker._get_setfile_path", return_value="/bin/echo"): preserve_file_attributes(src, dest) + @patch("subprocess.run") + def test_ffprobe_timeout(self, mock_run: MagicMock) -> None: + """Test ffprobe handles TimeoutExpired correctly.""" + from media_shrinker import probe_media, MediaShrinkerError + import subprocess + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=60) + with self.assertRaises(MediaShrinkerError) as ctx: + probe_media(Path("/dummy.mp4")) + + self.assertIn("ffprobe timed out for", str(ctx.exception)) + + @patch("subprocess.run") + def test_silencedetect_timeout(self, mock_run: MagicMock) -> None: + """Test silencedetect handles TimeoutExpired correctly.""" + from media_shrinker import detect_silence_intervals, MediaShrinkerError + import subprocess + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=3600) + with self.assertRaises(MediaShrinkerError) as ctx: + detect_silence_intervals(Path("/dummy.mp4")) + + self.assertIn("silencedetect timed out for", str(ctx.exception)) + + @patch("shutil.which", return_value="/usr/bin/brctl") + @patch("subprocess.run") + def test_brctl_timeout(self, mock_run: MagicMock, mock_which: MagicMock) -> None: + """Test brctl handles TimeoutExpired correctly.""" + from media_shrinker import download_from_icloud, MediaShrinkerError + import subprocess + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["brctl"], timeout=3600) + with self.assertRaises(MediaShrinkerError) as ctx: + download_from_icloud(Path("/dummy.mp4")) + + self.assertIn("iCloud download timed out for", str(ctx.exception)) + + @patch("media_shrinker._ensure_not_source_path") + @patch("media_shrinker._ensure_not_protected_source_path") + @patch("tempfile.mkstemp", return_value=(0, "/tmp/.test.tmp")) + @patch("os.close") + @patch("pathlib.Path.unlink") + @patch("subprocess.run") + def test_convert_timeout( + self, + mock_run: MagicMock, + mock_unlink: MagicMock, + mock_close: MagicMock, + mock_mkstemp: MagicMock, + mock_ensure_prot: MagicMock, + mock_ensure_src: MagicMock, + ) -> None: + """Test _execute_plan handles TimeoutExpired correctly.""" + from media_shrinker import _execute_plan, ConversionPlan, MediaShrinkerError + import subprocess + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=3600) + plan = ConversionPlan( + strategy="test", + input_path=Path("/dummy.mp4"), + output_path=Path("/tmp/dummy.mp4"), + ffmpeg_args=["-i", "{input}", "-c:a", "copy", "{output}"], + audio_bitrate_bps=128000 + ) + + with self.assertRaises(MediaShrinkerError) as ctx: + _execute_plan( + plan, source=Path("/dummy.mp4"), final_output=Path("/tmp/dummy.mp4"), overwrite=True, protected_sources=set(), ffmpeg_path="ffmpeg" + ) + + self.assertIn("ffmpeg timed out for", str(ctx.exception)) + + @patch("subprocess.run") + def test_copy_macos_creation_time_timeout(self, mock_run: MagicMock) -> None: + """Test _copy_macos_creation_time ignores TimeoutExpired.""" + from media_shrinker import _copy_macos_creation_time + import subprocess + + class MockStat: + st_birthtime = 123456789.0 + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["SetFile"], timeout=60) + # Should not raise + _copy_macos_creation_time(MockStat(), Path("/dest"), "SetFile")