From 2040839c8f67061c66a64a0b0912ffdaa305bd8e Mon Sep 17 00:00:00 2001 From: Yutsuo Date: Sun, 30 Aug 2026 13:56:52 -0300 Subject: [PATCH] fix: validate ffprobe output structure --- src/voice_transcriptor/services/media_probe.py | 16 +++++++++++++--- tests/test_media_probe.py | 6 ++++++ 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/src/voice_transcriptor/services/media_probe.py b/src/voice_transcriptor/services/media_probe.py index b952062..95923d8 100644 --- a/src/voice_transcriptor/services/media_probe.py +++ b/src/voice_transcriptor/services/media_probe.py @@ -50,18 +50,28 @@ def validate_media_path(path: Path) -> Path: def parse_probe_output(path: Path, size_bytes: int, payload: str) -> MediaInfo: try: - data: dict[str, Any] = json.loads(payload) + data: Any = json.loads(payload) except (json.JSONDecodeError, TypeError) as exc: raise ProbeOutputError("FFprobe returned invalid data.") from exc + if not isinstance(data, dict): + raise ProbeOutputError("FFprobe returned invalid data.") + format_data = data.get("format", {}) + streams = data.get("streams", []) + if ( + not isinstance(format_data, dict) + or not isinstance(streams, list) + or not all(isinstance(stream, dict) for stream in streams) + ): + raise ProbeOutputError("FFprobe returned invalid data.") duration: float | None = None - raw_duration = data.get("format", {}).get("duration") + raw_duration = format_data.get("duration") try: if raw_duration is not None: duration = float(raw_duration) except (TypeError, ValueError): duration = None codec = next( - (stream.get("codec_name") for stream in data.get("streams", []) if stream.get("codec_type") == "audio"), + (stream.get("codec_name") for stream in streams if stream.get("codec_type") == "audio"), None, ) return MediaInfo(path=path, size_bytes=size_bytes, duration_seconds=duration, audio_codec=codec) diff --git a/tests/test_media_probe.py b/tests/test_media_probe.py index a4144f2..8eead66 100644 --- a/tests/test_media_probe.py +++ b/tests/test_media_probe.py @@ -69,6 +69,12 @@ def test_rejects_malformed_probe_json(tmp_path: Path) -> None: parse_probe_output(tmp_path / "bad.mov", 0, "not json") +@pytest.mark.parametrize("payload", ["[]", '{"format": []}', '{"streams": [null]}']) +def test_rejects_structurally_invalid_probe_json(tmp_path: Path, payload: str) -> None: + with pytest.raises(ProbeOutputError, match="invalid data"): + parse_probe_output(tmp_path / "bad.mov", 0, payload) + + def test_rejects_missing_file(tmp_path: Path) -> None: with pytest.raises(MediaProbeError, match="does not exist"): validate_media_path(tmp_path / "missing.mov")