diff --git a/tests/tracks/test_encrypted_sample_entry.py b/tests/tracks/test_encrypted_sample_entry.py new file mode 100644 index 00000000..3874d2ce --- /dev/null +++ b/tests/tracks/test_encrypted_sample_entry.py @@ -0,0 +1,86 @@ +"""Self-check for ``has_encrypted_sample_entry``: a finalized MP4 whose moov still carries +an encv/enca sample entry means the decrypt tool skipped/failed the track (frma was never +applied), so the detector must flag it before the file reaches the muxer. Detection is +scoped to moov, so stray 4CC byte runs in mdat must never trip (or mask) it. + +Run: uv run pytest tests/tracks/test_encrypted_sample_entry.py +""" + +from __future__ import annotations + +import struct +from pathlib import Path + +from unshackle.core.tracks.track import has_encrypted_sample_entry + + +def _box(fmt: bytes, payload: bytes = b"") -> bytes: + return struct.pack(">I", 8 + len(payload)) + fmt + payload + + +def _sample_entry(fmt: bytes, payload: bytes = b"") -> bytes: + body = b"\x00" * 6 + struct.pack(">H", 1) + payload # reserved + data_reference_index + return struct.pack(">I", 8 + len(body)) + fmt + body + + +def _stsd(entries: list[bytes]) -> bytes: + body = b"\x00" * 4 + struct.pack(">I", len(entries)) + b"".join(entries) # version/flags + count + return _box(b"stsd", body) + + +def _moov(*children: bytes) -> bytes: + stbl = _box(b"stbl", b"".join(children)) + return _box(b"moov", _box(b"trak", _box(b"mdia", _box(b"minf", stbl)))) + + +FTYP = _box(b"ftyp", b"isom\x00\x00\x02\x00isomiso2") +# stray [size][encv] byte run as compressed A/V payload could randomly contain +STRAY_ENCV = b"\xde\xad" * 4 + struct.pack(">I", 100) + b"encv" + b"\xbe\xef" * 8 + + +def _write(tmp_path: Path, name: str, *boxes: bytes) -> Path: + f = tmp_path / name + f.write_bytes(b"".join(boxes)) + return f + + +def test_dual_stsd_encv_and_avc1_detected(tmp_path: Path) -> None: + sinf = _box(b"sinf", _box(b"frma", b"avc1")) + avc1 = _sample_entry(b"avc1", b"\x00" * 70) + encv = _sample_entry(b"encv", b"\x00" * 70 + sinf) + f = _write(tmp_path, "dual.mp4", FTYP, _moov(_stsd([avc1, encv])), _box(b"mdat", b"\x00" * 32)) + assert has_encrypted_sample_entry(f) is True + + +def test_single_avc1_not_detected(tmp_path: Path) -> None: + avc1 = _sample_entry(b"avc1", b"\x00" * 70) + f = _write(tmp_path, "clear.mp4", FTYP, _moov(_stsd([avc1])), _box(b"mdat", b"\x00" * 32)) + assert has_encrypted_sample_entry(f) is False + + +def test_stray_encv_in_mdat_not_detected(tmp_path: Path) -> None: + # clean moov, mdat happens to contain a plausible [size][encv] run -> must NOT flag + avc1 = _sample_entry(b"avc1", b"\x00" * 70) + f = _write(tmp_path, "stray.mp4", FTYP, _moov(_stsd([avc1])), _box(b"mdat", STRAY_ENCV)) + assert has_encrypted_sample_entry(f) is False + + +def test_leading_mdat_does_not_mask_real_encv(tmp_path: Path) -> None: + # progressive layout (mdat before moov) with a stray run must not hide a genuine encv + enca = _sample_entry(b"enca", b"\x00" * 20 + _box(b"sinf", _box(b"frma", b"mp4a"))) + f = _write(tmp_path, "masked.mp4", FTYP, _box(b"mdat", STRAY_ENCV), _moov(_stsd([enca]))) + assert has_encrypted_sample_entry(f) is True + + +def test_text_containing_enca_in_moov_not_detected(tmp_path: Path) -> None: + # "enca" inside a metadata string has no plausible box size before it -> must NOT flag + avc1 = _sample_entry(b"avc1", b"\x00" * 70) + udta = _box(b"udta", b"data is encapsulated here") + f = _write(tmp_path, "text.mp4", FTYP, _box(b"moov", _stsd([avc1]) + udta)) + assert has_encrypted_sample_entry(f) is False + + +def test_missing_or_bogus_file(tmp_path: Path) -> None: + assert has_encrypted_sample_entry(tmp_path / "nope.mp4") is False + bogus = _write(tmp_path, "bogus.bin", b"\x00\x01\x02\x03not-an-mp4") + assert has_encrypted_sample_entry(bogus) is False diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 708c76c3..6f2482f3 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -67,6 +67,7 @@ from unshackle.core.tracks import Audio, Subtitle, Tracks, Video from unshackle.core.tracks.attachment import Attachment from unshackle.core.tracks.dv_fixup import apply_dv_fixup from unshackle.core.tracks.hybrid import Hybrid +from unshackle.core.tracks.track import has_encrypted_sample_entry from unshackle.core.utilities import ( find_font_with_fallbacks, find_missing_langs, @@ -1895,6 +1896,12 @@ class dl: if track.needs_repack: track.repackage() events.emit(events.Types.TRACK_REPACKED, track=track) + # residual encrypted sample entry after repack => never decrypted + if has_encrypted_sample_entry(track.path): + self.log.warning( + f"Track {track.id} still has an encrypted sample entry after repacking, so " + "decryption likely failed and the muxed output may be unplayable." + ) media_info = MediaInfo.parse(track.path) media_infos[id(track)] = media_info @@ -3152,6 +3159,11 @@ class dl: vui_folded.add(id(track)) has_repacked = True events.emit(events.Types.TRACK_REPACKED, track=track) + if has_encrypted_sample_entry(track.path): + self.log.warning( + f"Track {track.id} still has an encrypted sample entry after repacking, so " + "decryption likely failed and the muxed output may be unplayable." + ) if has_repacked: # we don't want to fill up the log with "Repacked x track" self.log.info("Repacked one or more tracks with FFMPEG") diff --git a/unshackle/core/tracks/track.py b/unshackle/core/tracks/track.py index 30613a3f..53a74a0c 100644 --- a/unshackle/core/tracks/track.py +++ b/unshackle/core/tracks/track.py @@ -48,6 +48,58 @@ def direct_session(session: Union[Session, "RnetSession"]) -> Session: return new +def _read_top_level_box(path: Path, box_type: bytes) -> Optional[bytes]: + # seek through top-level box headers; only the wanted box's bytes are read into RAM + file_size = path.stat().st_size + with path.open("rb") as f: + start = 0 + while start + 8 <= file_size: + header = f.read(8) + if len(header) < 8: + return None + size = int.from_bytes(header[:4], "big") + if size == 1: # 64-bit largesize + large = f.read(8) + if len(large) < 8: + return None + size = int.from_bytes(large, "big") + elif size == 0: # box runs to EOF + size = file_size - start + if size < 8: + return None + if header[4:8] == box_type: + f.seek(start) + return f.read(size) + start += size + f.seek(start) + return None + + +def has_encrypted_sample_entry(path: Path) -> bool: + """True if the MP4's moov still carries an encrypted sample entry (encv/enca). + + A faithful decrypt rewrites encv/enca back to the real codec 4CC via frma, so a + survivor means the decrypt tool skipped/failed the track. This is a 4CC scan + scoped to the moov box (where sample entries live), not a structural parse; + scoping avoids chance byte collisions in mdat. Any read error -> False. + """ + try: + moov = _read_top_level_box(path, b"moov") + except Exception: + return False + if not moov: + return False + for tok in (b"encv", b"enca"): + i = moov.find(tok, 4) + while i != -1: + # require a plausible sample-entry box size right before the 4CC + size = int.from_bytes(moov[i - 4 : i], "big") + if 16 <= size <= len(moov) - (i - 4): + return True + i = moov.find(tok, i + 1) + return False + + @dataclass class DownloadContext: """Shared arguments passed to each manifest's ``download_track``.""" @@ -379,6 +431,10 @@ class Track: self.drm = None events.emit(events.Types.TRACK_DECRYPTED, track=self, drm=drm, segment=None) progress(downloaded="Decrypted", completed=100, total=100) + # residual encv/enca => decrypt tool skipped/failed; force a repack so the + # muxer isn't first-FourCC-locked to a generic (encrypted) codec. OR-in only. + if has_encrypted_sample_entry(save_path): + self.needs_repack = True if track_type == "Subtitle" and self.codec.name not in ("fVTT", "fTTML"): track_data = self.path.read_bytes()