Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 42 additions & 25 deletions media_shrinker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()}"
Expand Down Expand Up @@ -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()}"
Expand Down Expand Up @@ -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()}"
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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:
Expand Down
10 changes: 10 additions & 0 deletions tests/test_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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()
Loading