diff --git a/.jules/sentinel.md b/.jules/sentinel.md index d71c5b6..8f573fc 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -37,7 +37,20 @@ **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. +## 2026-06-25 - [Sentinel: Strix CI Command Injection False Positives] +**Vulnerability:** CI security scanners (like Strix) falsely reporting command injection vulnerabilities when `shell=False` is omitted. +**Learning:** Some static analysis security tools flag `subprocess.run` calls as vulnerable to command injection if the `shell` argument is missing, even when the command is passed safely as a list of strings. +**Prevention:** Explicitly include `shell=False` in all `subprocess.run` calls, even when passing arguments as a list, to prevent false positive command injection alerts from CI security scanners like Strix. + ## 2026-07-05 - [Sentinel: Fix Argument Injection Vulnerability] **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-27 - Uncontrolled Resource Consumption (DoS) via Subprocess Timeouts +**Vulnerability:** External binaries (`ffmpeg`, `ffprobe`, `brctl`, `SetFile`) were executed via `subprocess.run` without an explicit `timeout` parameter. If these external processes hung indefinitely (e.g., waiting for I/O, hitting a bug, or processing a malicious input file that triggers an infinite loop), the application thread would block forever. +**Learning:** Python's `subprocess.run` blocks until the child process completes. Without a timeout, a blocked child process translates directly into a blocked application thread, leading to Uncontrolled Resource Consumption (CWE-400) and potential Denial of Service (DoS). This is especially critical in web applications or batch processing systems where thread exhaustion can take down the entire service. +**Prevention:** Always configure an explicit `timeout` parameter tailored to the specific binary (e.g., 60s for quick probes, 3600s+ for heavy processing) when calling `subprocess.run`. Catch `subprocess.TimeoutExpired` exceptions to handle the failure gracefully. Python will automatically terminate the child process if the timeout is reached. +## 2024-05-28 - Uncontrolled Resource Consumption (DoS) via Subprocess Timeouts +**Vulnerability:** External binaries (`ffmpeg`, `ffprobe`, `brctl`, `SetFile`) were executed via `subprocess.run` without an explicit `timeout` parameter. If these external processes hung indefinitely (e.g., waiting for I/O, hitting a bug, or processing a malicious input file that triggers an infinite loop), the application thread would block forever. +**Learning:** Python's `subprocess.run` blocks until the child process completes. Without a timeout, a blocked child process translates directly into a blocked application thread, leading to Uncontrolled Resource Consumption (CWE-400) and potential Denial of Service (DoS). This is especially critical in web applications or batch processing systems where thread exhaustion can take down the entire service. +**Prevention:** Always configure an explicit `timeout` parameter tailored to the specific binary (e.g., 60s for quick probes, 3600s+ for heavy processing) when calling `subprocess.run`. Catch `subprocess.TimeoutExpired` exceptions to handle the failure gracefully. Python will automatically terminate the child process if the timeout is reached. diff --git a/media_shrinker.py b/media_shrinker.py index 7f38dd7..8b86b4b 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -8,6 +8,7 @@ """ import argparse +import logging import functools from concurrent.futures import ThreadPoolExecutor, as_completed import json @@ -88,6 +89,11 @@ SILENCE_END_RE = re.compile(r"silence_end:\s*(?P[0-9]+(?:\.[0-9]+)?)") +PROBE_TIMEOUT_SECONDS = 60.0 +PROCESS_TIMEOUT_SECONDS = 3600.0 + +logger = logging.getLogger(__name__) + class MediaShrinkerError(RuntimeError): """Raised when a media file cannot be processed safely.""" @@ -465,7 +471,13 @@ def probe_media( "-i", str(Path(source_path).resolve()), ] - completed = subprocess.run(command, check=False, capture_output=True, text=True) + try: + completed = subprocess.run( + command, check=False, capture_output=True, text=True, shell=False, timeout=PROBE_TIMEOUT_SECONDS + ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffprobe exceeded {PROBE_TIMEOUT_SECONDS}s timeout for {source_path}") from exc + if completed.returncode != 0: raise MediaShrinkerError( f"ffprobe failed for {source_path}: {completed.stderr.strip()}" @@ -518,17 +530,23 @@ 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, + shell=False, + timeout=PROCESS_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"silencedetect exceeded {PROCESS_TIMEOUT_SECONDS}s timeout for {source_path}") from exc + if completed.returncode != 0: raise MediaShrinkerError( f"silencedetect failed for {source_path}: {completed.stderr.strip()}" @@ -643,12 +661,18 @@ 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, + shell=False, + timeout=PROCESS_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"iCloud download exceeded {PROCESS_TIMEOUT_SECONDS}s timeout for {source_path}") from exc + if completed.returncode != 0: raise MediaShrinkerError( f"iCloud download failed for {source_path}: {completed.stderr.strip()}" @@ -1586,10 +1610,12 @@ def _execute_plan( ) try: completed = subprocess.run( - command, check=False, capture_output=True, text=True + command, check=False, capture_output=True, text=True, shell=False, timeout=PROCESS_TIMEOUT_SECONDS ) except FileNotFoundError as exc: raise MediaShrinkerError(f"ffmpeg not found: {ffmpeg_path}") from exc + except subprocess.TimeoutExpired as exc: + raise MediaShrinkerError(f"ffmpeg exceeded {PROCESS_TIMEOUT_SECONDS}s timeout for {source}") from exc if completed.returncode != 0: raise MediaShrinkerError( @@ -1654,12 +1680,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, - ) + try: + subprocess.run( + [setfile_path, "-d", creation_date, str(dest.resolve())], + check=False, + capture_output=True, + text=True, + shell=False, + timeout=PROBE_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: # pragma: no cover + logger.warning(f"SetFile exceeded {PROBE_TIMEOUT_SECONDS}s timeout copying macOS timestamps for {dest}") def _format_result(root: Path, result: ConversionResult) -> str: diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 0ab2397..81f9ea8 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -1,3 +1,4 @@ +import subprocess import json import os import tempfile @@ -1914,3 +1915,34 @@ 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_subprocess_timeout_expired_handled(self, mock_run): + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=60.0) + with self.assertRaises(media_shrinker.MediaShrinkerError): + media_shrinker.probe_media("dummy.mp4") + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=3600.0) + with self.assertRaises(media_shrinker.MediaShrinkerError): + media_shrinker.detect_silence_intervals("dummy.mp4") + + with patch("media_shrinker.shutil.which", return_value="brctl"): + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["brctl"], timeout=3600.0) + with self.assertRaises(media_shrinker.MediaShrinkerError): + media_shrinker.download_from_icloud(Path("dummy.mp4"), brctl_path="brctl") + + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["SetFile"], timeout=60.0) + # Should not raise + media_shrinker._copy_macos_creation_time(os.stat_result((1, 2, 3, 4, 5, 6, 7, 8, 9, 10)), Path("dest.mp4"), "SetFile") + + @patch("media_shrinker.subprocess.run") + def test_execute_plan_timeout_expired(self, mock_run): + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=3600.0) + plan = media_shrinker.ConversionPlan( + strategy="test", + input_path=Path("input.mp4"), + output_path=Path("output.mp4"), + ffmpeg_args=["-i", "input.mp4", "output.mp4"] + ) + with self.assertRaises(media_shrinker.MediaShrinkerError): + media_shrinker._execute_plan(plan, Path("input.mp4"), Path("output.mp4"), ffmpeg_path="ffmpeg", overwrite=True)