diff --git a/media_shrinker.py b/media_shrinker.py index 1cc2a9f..41c9122 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=30.0) + 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=180.0, + ) + 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=300.0, + ) + 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()}" @@ -1550,8 +1561,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=1800.0 ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffmpeg conversion timed out for {source}") from exc except FileNotFoundError as exc: raise MediaShrinkerError(f"ffmpeg not found: {ffmpeg_path}") from exc @@ -1618,12 +1631,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=30.0, + ) + except subprocess.TimeoutExpired: + pass def _format_result(root: Path, result: ConversionResult) -> str: diff --git a/tests/test_security.py b/tests/test_security.py index 6ec6aff..af1edef 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -2,6 +2,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch +import subprocess from media_shrinker import build_silencedetect_command, MediaShrinkerError, probe_media class SecurityTests(unittest.TestCase): @@ -33,5 +34,14 @@ def test_probe_media_uses_explicit_input_flag_for_dash_prefixed_path( input_index = command.index("-i") self.assertEqual(command[input_index + 1], str(source_path)) + @patch("media_shrinker.subprocess.run") + def test_probe_media_enforces_timeout(self, mock_run: MagicMock): + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=30.0) + with patch.object(Path, "stat") as mock_stat: + mock_stat.return_value = MagicMock(st_size=10) + with self.assertRaises(MediaShrinkerError) as cm: + probe_media(Path("test.wav")) + self.assertIn("timed out", str(cm.exception)) + if __name__ == "__main__": unittest.main()