feat(drm): auto-repack tracks with a residual encrypted sample entry

Detect an encv/enca sample entry surviving decryption (moov-scoped 4CC scan) and force a repack so mkvmerge isn't first-FourCC-locked to a generic encrypted codec; warn if one persists after repacking.

Closes #142

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

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")

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()