fix(media): restrict FFmpeg/FFprobe input protocols

.jules/sentinel.md

@@ -17,10 +17,10 @@
 **Learning:** In FastAPI/Starlette, `file.filename` can be unsafe or empty. Using `Path(file.filename).name` may resolve to `.` or `..`, leading to OS-level exceptions when attempting to write data. If resource allocation (like `tempfile.mkdtemp()`) occurs outside the scope of the `try...finally` (or `BackgroundTasks` cleanup) that handles these errors, an attacker can intentionally leak resources by sending manipulated paths.
 **Prevention:** Always place resource allocation inside or immediately before the associated `try...finally` block. Sanitize and validate filenames retrieved from `UploadFile.filename` by ensuring they are non-empty and are not relative references (`.` or `..`), providing a safe default fallback.
 
-## 2026-06-07 - FFmpeg SSRF/LFI Vulnerability Fix
-**Vulnerability:** Local File Inclusion and Server-Side Request Forgery via unrestricted FFmpeg/FFprobe protocols.
-**Learning:** The application executed FFmpeg and FFprobe on user-supplied media files without protocol restrictions. Malicious files (like HLS playlists) could leverage protocols like `http` to exfiltrate data or access internal services.
-**Prevention:** Always enforce `"-protocol_whitelist", "file,crypto,data"` before the input flag when invoking FFmpeg/FFprobe to restrict processing to safe local protocols.
+## 2026-06-07 - FFmpeg SSRF Vulnerability Fix
+**Vulnerability:** Server-Side Request Forgery via unrestricted FFmpeg/FFprobe protocols.
+**Learning:** The application executed FFmpeg and FFprobe on user-supplied media files without protocol restrictions. Malicious files (like HLS playlists) could leverage protocols like `http` to exfiltrate data or access internal network services.
+**Prevention:** Always enforce `"-protocol_whitelist", "file,crypto,data"` before the input flag when invoking FFmpeg/FFprobe to restrict processing to safe local protocols and prevent network SSRF. Note: This does not prevent Local File Inclusion (LFI) via the `file` protocol.
 
 ## 2026-06-09 - [Sentinel: FFmpeg Argument Injection Vulnerability Fix]
 **Vulnerability:** Argument injection via maliciously crafted filenames.

audio_library.py

@@ -2072,6 +2072,8 @@ def audio_duration_seconds(
             "format=duration",
             "-of",
             "default=noprint_wrappers=1:nokey=1",
+            "-protocol_whitelist",
+            "file,crypto,data",
             media_input,
         ]
         completed = subprocess.run(
@@ -2861,7 +2863,7 @@ def decode_audio_for_mlx(
             # Input-side seeking avoids decoding every earlier chunk; ffmpeg's
             # default accurate_seek still discards samples before this boundary.
             command.extend(("-ss", f"{start_seconds:.6f}"))
-        command.extend(("-i", media_input))
+        command.extend(("-protocol_whitelist", "file,crypto,data", "-i", media_input))
         if duration_seconds is not None:
             command.extend(("-t", f"{duration_seconds:.6f}"))
         command.extend(
@@ -2941,6 +2943,8 @@ def detect_silence_intervals(
     command = [
         str(ffmpeg),
         "-nostdin",
+        "-protocol_whitelist",
+        "file,crypto,data",
         "-i",
         media_input,
         "-af",

tests/test_audio_library.py

@@ -4529,12 +4529,14 @@ def __truediv__(self, _value):
             )
         command = run.call_args.args[0]
         self.assertEqual(
-            command[:8],
+            command[:10],
             [
                 "/usr/bin/ffmpeg",
                 "-nostdin",
                 "-ss",
                 "299.000000",
+                "-protocol_whitelist",
+                "file,crypto,data",
                 "-i",
                 "recording.wav",
                 "-t",
@@ -4571,7 +4573,7 @@ def __truediv__(self, _value):
         ):
             self.assertEqual(audio_library.decode_audio_for_mlx(artifact), "decoded")
         descriptor = handle.fileno()
-        self.assertEqual(run.call_args.args[0][3], f"/dev/fd/{descriptor}")
+        self.assertEqual(run.call_args.args[0][5], f"/dev/fd/{descriptor}")
         self.assertEqual(run.call_args.kwargs["pass_fds"], (descriptor,))
         self.assertNotIn("stdin", run.call_args.kwargs)
         handle.close()

tests/test_ffmpeg_protocol_whitelist_contract.py

@@ -0,0 +1,116 @@
+"""Executable contracts for FFmpeg/FFprobe input protocol admission."""
+
+from __future__ import annotations
+
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import audio_library
+
+
+class FfmpegProtocolWhitelistContractTests(unittest.TestCase):
+    """Keep untrusted media parsing on the reviewed local-protocol boundary."""
+
+    _ALLOWED_PROTOCOLS = {"file", "crypto", "data"}
+    _NETWORK_PROTOCOLS = {
+        "ftp",
+        "gopher",
+        "http",
+        "https",
+        "rtmp",
+        "rtsp",
+        "sftp",
+        "smb",
+        "srt",
+        "tcp",
+        "udp",
+    }
+
+    def assert_local_input_protocols(self, command: list[str]) -> int:
+        """Return the whitelist index after checking its exact network boundary."""
+
+        whitelist_index = command.index("-protocol_whitelist")
+        protocols = set(command[whitelist_index + 1].split(","))
+        self.assertEqual(protocols, self._ALLOWED_PROTOCOLS)
+        self.assertTrue(protocols.isdisjoint(self._NETWORK_PROTOCOLS))
+        return whitelist_index
+
+    def test_ffprobe_duration_applies_whitelist_immediately_before_input(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=[], returncode=0, stdout="12.5\n", stderr=""
+        )
+        with tempfile.TemporaryDirectory() as root:
+            media_path = Path(root) / "recording.m4a"
+            media_path.write_bytes(b"fixture")
+            with (
+                patch(
+                    "audio_library.trusted_ffprobe_binary",
+                    return_value=Path("/usr/bin/ffprobe"),
+                ),
+                patch("audio_library.subprocess.run", return_value=completed) as run,
+            ):
+                self.assertEqual(audio_library.audio_duration_seconds(media_path), 12.5)
+
+        command = run.call_args.args[0]
+        whitelist_index = self.assert_local_input_protocols(command)
+        self.assertEqual(command[whitelist_index + 2], str(media_path))
+        self.assertEqual(run.call_args.kwargs["pass_fds"], ())
+
+    def test_silence_detection_applies_whitelist_before_input_flag(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=[], returncode=0, stdout=b"", stderr=b""
+        )
+        with (
+            patch(
+                "audio_library.trusted_ffmpeg_binary",
+                return_value=Path("/usr/bin/ffmpeg"),
+            ),
+            patch("audio_library.subprocess.run", return_value=completed) as run,
+        ):
+            self.assertEqual(
+                audio_library.detect_silence_intervals(Path("recording.m4a")), []
+            )
+
+        command = run.call_args.args[0]
+        whitelist_index = self.assert_local_input_protocols(command)
+        self.assertEqual(
+            command[whitelist_index + 2 : whitelist_index + 4],
+            ["-i", "recording.m4a"],
+        )
+        self.assertEqual(run.call_args.kwargs["pass_fds"], ())
+
+
+
+    def test_decode_mlx_audio_applies_whitelist_before_input_flag(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=[], returncode=0, stdout=b"dummy", stderr=b""
+        )
+        with tempfile.TemporaryDirectory() as root:
+            media_path = Path(root) / "recording.m4a"
+            media_path.write_bytes(b"fixture")
+            with (
+                patch(
+                    "audio_library.trusted_ffmpeg_binary",
+                    return_value=Path("/usr/bin/ffmpeg"),
+                ),
+                patch("audio_library.subprocess.run", return_value=completed) as run,
+            ):
+                with patch.dict("sys.modules", {"mlx": unittest.mock.MagicMock(), "mlx.core": unittest.mock.MagicMock(), "numpy": unittest.mock.MagicMock()}):
+                    try:
+                        audio_library.decode_audio_for_mlx(media_path)
+                    except Exception:
+                        pass
+
+        command = run.call_args.args[0]
+        whitelist_index = self.assert_local_input_protocols(command)
+        self.assertEqual(
+            command[whitelist_index + 2 : whitelist_index + 4],
+            ["-i", str(media_path)],
+        )
+        self.assertEqual(run.call_args.kwargs["pass_fds"], ())
+
+if __name__ == '__main__':
+    unittest.main()