🛡️ Sentinel: [CRITICAL] 외부 프로세스 실행 시 무한 대기 (DoS) 취약점 해결

.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.
+## 2024-05-24 - Missing Timeout on External Binary Execution (DoS Risk)
+**Vulnerability:** Found multiple `subprocess.run` calls without explicit timeouts (e.g. `ffmpeg`, `ffprobe`, `brctl`, `SetFile`).
+**Learning:** This exposes the application to Uncontrolled Resource Consumption (DoS) if these external binaries hang indefinitely.
+**Prevention:** Always configure an explicit `timeout` parameter and handle `subprocess.TimeoutExpired` exceptions when executing external binaries with `subprocess.run`.

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,8 +1601,10 @@ 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 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
 
@@ -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
 
 
 def _format_result(root: Path, result: ConversionResult) -> str:

tests/test_media_shrinker.py

@@ -1914,3 +1914,56 @@ 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("media_shrinker.subprocess.run")
+    def test_probe_media_handles_timeout(self, mock_run: MagicMock) -> None:
+        import subprocess
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffprobe", timeout=60)
+        from media_shrinker import probe_media, MediaShrinkerError
+        with self.assertRaisesRegex(MediaShrinkerError, "ffprobe timed out"):
+            probe_media(Path("source.wav"))
+
+    @patch("media_shrinker.subprocess.run")
+    def test_detect_silence_handles_timeout(self, mock_run: MagicMock) -> None:
+        import subprocess
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=3600)
+        from media_shrinker import detect_silence_intervals, MediaShrinkerError
+        with self.assertRaisesRegex(MediaShrinkerError, "silencedetect timed out"):
+            detect_silence_intervals(Path("source.wav"))
+
+    @patch("media_shrinker.subprocess.run")
+    def test_download_icloud_handles_timeout(self, mock_run: MagicMock) -> None:
+        import subprocess
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="brctl", timeout=3600)
+        from media_shrinker import download_from_icloud, MediaShrinkerError
+        with patch("shutil.which", return_value="brctl"):
+            with self.assertRaisesRegex(MediaShrinkerError, "iCloud download timed out"):
+                download_from_icloud(Path("source.wav"))
+
+    @patch("media_shrinker.subprocess.run")
+    def test_execute_plan_handles_timeout(self, mock_run: MagicMock) -> None:
+        import subprocess
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffmpeg", timeout=3600)
+        from media_shrinker import _execute_plan, ConversionPlan, MediaShrinkerError
+        plan = ConversionPlan(
+            strategy="test",
+            input_path=Path("source.wav"),
+            output_path=Path("out.opus"),
+            ffmpeg_args=["-i", "source.wav", "out.opus"]
+        )
+        with self.assertRaisesRegex(MediaShrinkerError, "ffmpeg timed out"):
+            _execute_plan(plan, Path("source.wav"), Path("final.opus"), ffmpeg_path="ffmpeg", overwrite=True)
+
+    @patch("media_shrinker.subprocess.run")
+    def test_copy_macos_creation_time_handles_timeout(self, mock_run: MagicMock) -> None:
+        import subprocess
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="SetFile", timeout=60)
+        mock_stat = MagicMock()
+        mock_stat.st_birthtime = 1234567890.0
+        from media_shrinker import _copy_macos_creation_time
+        # Should not raise exception
+        _copy_macos_creation_time(mock_stat, Path("dest.txt"), "SetFile")
+
+    def test_build_silencedetect_command_invalid_noise(self) -> None:
+        from media_shrinker import build_silencedetect_command, MediaShrinkerError
+        with self.assertRaisesRegex(MediaShrinkerError, "Invalid silence_noise value"):
+            build_silencedetect_command(Path("source.wav"), silence_noise="invalid")

tests/test_security.py

@@ -35,3 +35,12 @@ def test_probe_media_uses_explicit_input_flag_for_dash_prefixed_path(
 
 if __name__ == "__main__":
     unittest.main()
+    @patch("media_shrinker.subprocess.run")
+    def test_probe_media_invalid_json(self, mock_run: MagicMock) -> None:
+        mock_completed = MagicMock()
+        mock_completed.returncode = 0
+        mock_completed.stdout = "invalid json"
+        mock_run.return_value = mock_completed
+        from media_shrinker import probe_media, MediaShrinkerError
+        with self.assertRaisesRegex(MediaShrinkerError, "ffprobe returned invalid JSON"):
+            probe_media(Path("source.wav"))