🛡️ Sentinel: [MEDIUM] Fix missing timeout configurations

.gitignore

@@ -2,3 +2,4 @@ __pycache__/
 *.py[cod]
 .DS_Store
 .coverage
+venv/

.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.

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:

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")