🛡️ Sentinel: [MEDIUM] DoS 방지 - subprocess timeout 추가

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

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:

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