From 762980d1ec9be37186f640d8fb189d0b284538f5 Mon Sep 17 00:00:00 2001 From: Hugoved Date: Fri, 24 Apr 2026 11:56:40 +0200 Subject: [PATCH] Add files via upload --- pymp4dump.py | 1092 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1092 insertions(+) create mode 100644 pymp4dump.py diff --git a/pymp4dump.py b/pymp4dump.py new file mode 100644 index 0000000..14b7793 --- /dev/null +++ b/pymp4dump.py @@ -0,0 +1,1092 @@ +from __future__ import annotations +import argparse +import base64 +import binascii +import json +import os +import struct +import sys +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, List, Optional, Tuple + +VERSION = "1.0" +BANNER = "MP4 File Dumper - Python Version %s" % VERSION +MAX_SCAN_BOXES = 1_000_000 +MAX_PREVIEW_BYTES = 64 +CONTAINER_BOXES = { + "moov", "trak", "mdia", "minf", "dinf", "stbl", "edts", "udta", "meta", "ilst", + "moof", "traf", "mfra", "skip", "wide", "ipro", "sinf", "schi", "tref", "iref", + "meco", "mere", "strk", "stri", "strd", "sbgp", "sgpd", "hnti", "hinf", "avcn", + "avss", "wave", "gmhd", "clip", "matt", "rmra", "rmda", "rmdb", "cmov", "cmvd", + "dref", "paen", "font", "vttc", "vtte", "vtta", "vttC", "stpp", "uuid", +} +FULL_BOX_TYPES = { + "mvhd", "tkhd", "mdhd", "hdlr", "vmhd", "smhd", "hmhd", "nmhd", "dref", "stsd", "stts", "ctts", + "cslg", "stsc", "stsz", "stz2", "stco", "co64", "stss", "stsh", "stdp", "sdtp", "stdp", "padb", + "elst", "mehd", "trex", "mfhd", "tfhd", "trun", "tfdt", "mfra", "tfra", "mfro", "pdin", "ftyp", + "mvex", "url ", "urn ", "meta", "pitm", "iloc", "iinf", "infe", "iprp", "ipco", "ipma", "ipro", + "sidx", "ssix", "pssh", "tenc", "schm", "frma", "saiz", "saio", "senc", "sgpd", "sbgp", "prft", +} +HANDLER_NAMES = { + "vide": "Video Track", "soun": "Audio Track", "hint": "Hint Track", "meta": "Metadata Track", + "text": "Text Track", "subt": "Subtitle Track", "sbtl": "Subtitle Track", "clcp": "Closed Caption", + "mdir": "Metadata", "tmcd": "Timecode Track", "auxv": "Auxiliary Video", "pict": "Picture Track", +} +PROTECTION_SYSTEM_IDS = { + "edef8ba979d64acea3c827dcd51d21ed": "Widevine", + "9a04f07998404286ab92e65be0885f95": "PlayReady", + "94ce86fb07ff4f43adb893d2fa968ca2": "FairPlay", + "1077efecc0b24d02ace33c1e52e2fb4b": "Common PSSH", +} + +class Mp4DumpError(Exception): + pass + +class Reader: + def __init__(self, data: bytes, base_offset: int = 0) -> None: + self.data = data + self.pos = 0 + self.base_offset = base_offset + + def remaining(self) -> int: + return len(self.data) - self.pos + + def tell(self) -> int: + return self.pos + + def seek(self, pos: int) -> None: + if pos < 0 or pos > len(self.data): + raise Mp4DumpError("seek outside buffer") + self.pos = pos + + def skip(self, size: int) -> None: + self.seek(self.pos + size) + + def read(self, size: int) -> bytes: + if size < 0 or self.pos + size > len(self.data): + raise Mp4DumpError("unexpected end of data") + out = self.data[self.pos:self.pos + size] + self.pos += size + return out + + def u8(self) -> int: + return self.read(1)[0] + + def i8(self) -> int: + return struct.unpack(">b", self.read(1))[0] + + def u16(self) -> int: + return struct.unpack(">H", self.read(2))[0] + + def i16(self) -> int: + return struct.unpack(">h", self.read(2))[0] + + def u24(self) -> int: + b = self.read(3) + return (b[0] << 16) | (b[1] << 8) | b[2] + + def u32(self) -> int: + return struct.unpack(">I", self.read(4))[0] + + def i32(self) -> int: + return struct.unpack(">i", self.read(4))[0] + + def u64(self) -> int: + return struct.unpack(">Q", self.read(8))[0] + + def i64(self) -> int: + return struct.unpack(">q", self.read(8))[0] + + def fixed16_16(self) -> float: + return self.i32() / 65536.0 + + def fixed8_8(self) -> float: + return self.i16() / 256.0 + + def cstring(self) -> str: + start = self.pos + while self.pos < len(self.data) and self.data[self.pos] != 0: + self.pos += 1 + raw = self.data[start:self.pos] + if self.pos < len(self.data): + self.pos += 1 + return decode_text(raw) + + +def fourcc(raw: bytes) -> str: + return "".join(chr(b) if 32 <= b <= 126 else "." for b in raw) + + +def decode_text(raw: bytes) -> str: + for enc in ("utf-8", "utf-16-be", "latin-1"): + try: + return raw.decode(enc).rstrip("\x00") + except UnicodeDecodeError: + pass + return raw.hex() + + +def hex_preview(raw: bytes, limit: int = MAX_PREVIEW_BYTES) -> str: + if len(raw) <= limit: + return raw.hex() + return raw[:limit].hex() + "..." + + +def language_from_code(code: int) -> str: + chars = [] + for shift in (10, 5, 0): + chars.append(chr(((code >> shift) & 0x1F) + 0x60)) + return "".join(chars) + + +def qt_time(seconds_since_1904: int) -> str: + # Keep numeric output deterministic; avoid timezone/date dependencies. + return str(seconds_since_1904) + +@dataclass +class Box: + type: str + size: int + start: int + header_size: int + payload_start: int + payload_size: int + uuid: Optional[str] = None + version: Optional[int] = None + flags: Optional[int] = None + fields: Dict[str, Any] = field(default_factory=dict) + children: List["Box"] = field(default_factory=list) + error: Optional[str] = None + + @property + def end(self) -> int: + return self.start + self.size + + def to_json(self) -> Dict[str, Any]: + obj: Dict[str, Any] = { + "name": self.type, + "type": self.type, + "offset": self.start, + "size": self.size, + "header_size": self.header_size, + } + if self.uuid: + obj["uuid"] = self.uuid + if self.version is not None: + obj["version"] = self.version + if self.flags is not None: + obj["flags"] = self.flags + if self.fields: + obj.update(self.fields) + if self.error: + obj["error"] = self.error + if self.children: + obj["children"] = [child.to_json() for child in self.children] + return obj + +class Mp4Parser: + def __init__(self, data: bytes, verbosity: int = 0) -> None: + self.data = data + self.verbosity = max(0, min(3, verbosity)) + self.track_context: Dict[int, Dict[str, Any]] = {} + self.current_track_id: Optional[int] = None + self.current_handler: Optional[str] = None + self.current_timescale: Optional[int] = None + + def parse(self) -> List[Box]: + return self._parse_children(0, len(self.data), None) + + def _parse_children(self, start: int, end: int, parent_type: Optional[str]) -> List[Box]: + boxes: List[Box] = [] + pos = start + count = 0 + while pos + 8 <= end and count < MAX_SCAN_BOXES: + count += 1 + box = self._parse_box(pos, end, parent_type) + boxes.append(box) + if box.size <= 0: + break + next_pos = box.end + if next_pos <= pos: + break + pos = next_pos + return boxes + + def _parse_box(self, start: int, parent_end: int, parent_type: Optional[str]) -> Box: + r = Reader(self.data[start:parent_end], base_offset=start) + try: + small_size = r.u32() + typ = fourcc(r.read(4)) + header_size = 8 + size = small_size + if small_size == 1: + size = r.u64() + header_size = 16 + elif small_size == 0: + size = parent_end - start + uuid_value = None + if typ == "uuid": + uuid_value = r.read(16).hex() + header_size += 16 + if size < header_size: + size = header_size + if start + size > parent_end: + size = parent_end - start + payload_start = start + header_size + payload_size = max(0, size - header_size) + box = Box(typ, size, start, header_size, payload_start, payload_size, uuid_value) + self._decode_box(box, parent_type) + return box + except Exception as exc: + return Box("????", max(8, parent_end - start), start, 8, start + 8, max(0, parent_end - start - 8), error=str(exc)) + + def _decode_box(self, box: Box, parent_type: Optional[str]) -> None: + payload = self.data[box.payload_start:box.payload_start + box.payload_size] + r = Reader(payload, base_offset=box.payload_start) + saved_track = self.current_track_id + saved_handler = self.current_handler + saved_timescale = self.current_timescale + try: + decoder = getattr(self, "decode_" + box.type.strip().replace(" ", "_"), None) + if box.type == "url ": + decoder = self.decode_url + elif box.type == "urn ": + decoder = self.decode_urn + elif box.type == "©nam": + decoder = self.decode_data_container + if decoder: + decoder(box, r) + elif self._looks_like_full_box(box.type, payload): + self._read_full_box(box, r) + if self._is_container(box, parent_type): + child_start = box.payload_start + self._container_payload_skip(box) + if child_start < box.end: + box.children = self._parse_children(child_start, box.end, box.type) + except Exception as exc: + box.error = str(exc) + finally: + if box.type == "trak": + self.current_track_id = saved_track + self.current_handler = saved_handler + self.current_timescale = saved_timescale + + def _looks_like_full_box(self, typ: str, payload: bytes) -> bool: + return typ in FULL_BOX_TYPES and len(payload) >= 4 + + def _read_full_box(self, box: Box, r: Reader) -> None: + box.version = r.u8() + box.flags = r.u24() + + def _is_container(self, box: Box, parent_type: Optional[str]) -> bool: + if box.type in {"mdat", "free", "ftyp", "styp", "sidx", "ssix", "pssh", "emsg"}: + return False + if box.type in CONTAINER_BOXES: + return True + if parent_type in {"stsd", "ipco", "ilst", "wave", "sinf", "schi", "dref"}: + return box.payload_size >= 8 + return False + + def _container_payload_skip(self, box: Box) -> int: + if box.type == "meta": + return 4 if box.version is not None else 0 + if box.type in {"stsd", "dref", "ipro"}: + return 8 + return 0 + + def _register(self, key: str, value: Any) -> None: + if self.current_track_id is None: + return + ctx = self.track_context.setdefault(self.current_track_id, {}) + ctx[key] = value + + def decode_ftyp(self, box: Box, r: Reader) -> None: + box.fields["major_brand"] = fourcc(r.read(4)) + box.fields["minor_version"] = r.u32() + brands = [] + while r.remaining() >= 4: + brands.append(fourcc(r.read(4))) + box.fields["compatible_brands"] = brands + + decode_styp = decode_ftyp + + def decode_mvhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if box.version == 1: + creation, modification, timescale, duration = r.u64(), r.u64(), r.u32(), r.u64() + else: + creation, modification, timescale, duration = r.u32(), r.u32(), r.u32(), r.u32() + box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "timescale": timescale, "duration": duration}) + if r.remaining() >= 8: + box.fields["duration_seconds"] = duration / timescale if timescale else None + box.fields["rate"] = r.fixed16_16() + box.fields["volume"] = r.fixed8_8() + if r.remaining() >= 10: + r.skip(10) + if r.remaining() >= 36: + box.fields["matrix"] = [r.i32() for _ in range(9)] + if r.remaining() >= 24: + r.skip(24) + if r.remaining() >= 4: + box.fields["next_track_id"] = r.u32() + + def decode_tkhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if box.version == 1: + creation, modification, track_id, reserved, duration = r.u64(), r.u64(), r.u32(), r.u32(), r.u64() + else: + creation, modification, track_id, reserved, duration = r.u32(), r.u32(), r.u32(), r.u32(), r.u32() + self.current_track_id = track_id + self.track_context.setdefault(track_id, {"id": track_id}) + box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "track_id": track_id, "duration": duration}) + if r.remaining() >= 8: + r.skip(8) + if r.remaining() >= 8: + box.fields["layer"] = r.i16(); box.fields["alternate_group"] = r.i16(); box.fields["volume"] = r.fixed8_8(); r.skip(2) + if r.remaining() >= 36: + box.fields["matrix"] = [r.i32() for _ in range(9)] + if r.remaining() >= 8: + box.fields["width"] = r.fixed16_16(); box.fields["height"] = r.fixed16_16() + + def decode_mdhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if box.version == 1: + creation, modification, timescale, duration = r.u64(), r.u64(), r.u32(), r.u64() + else: + creation, modification, timescale, duration = r.u32(), r.u32(), r.u32(), r.u32() + lang = language_from_code(r.u16()) if r.remaining() >= 2 else "und" + if r.remaining() >= 2: + r.u16() + self.current_timescale = timescale + self._register("timescale", timescale) + box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "timescale": timescale, "duration": duration, "language": lang}) + box.fields["duration_seconds"] = duration / timescale if timescale else None + + def decode_hdlr(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 4: + box.fields["pre_defined"] = r.u32() + handler = fourcc(r.read(4)) if r.remaining() >= 4 else "" + self.current_handler = handler + self._register("handler", handler) + box.fields["handler_type"] = handler + box.fields["handler_name"] = HANDLER_NAMES.get(handler, "") + if r.remaining() >= 12: + r.skip(12) + if r.remaining() > 0: + box.fields["name"] = r.cstring() + + def decode_vmhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 8: + box.fields["graphics_mode"] = r.u16(); box.fields["op_color"] = [r.u16(), r.u16(), r.u16()] + + def decode_smhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 4: + box.fields["balance"] = r.fixed8_8(); r.skip(2) + + def decode_hmhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 8: + box.fields.update({"max_pdu_size": r.u16(), "avg_pdu_size": r.u16(), "max_bitrate": r.u32(), "avg_bitrate": r.u32()}) + + def decode_nmhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + + def decode_stsd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 4: + box.fields["entry_count"] = r.u32() + + def _decode_sample_entry_common(self, box: Box, r: Reader) -> None: + if r.remaining() >= 6: + r.skip(6) + if r.remaining() >= 2: + box.fields["data_reference_index"] = r.u16() + + def decode_avc1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_avc3(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_hev1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_hvc1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_dvhe(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_dvh1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_mp4v(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + def decode_encv(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r) + + def _decode_visual_sample_entry(self, box: Box, r: Reader) -> None: + self._decode_sample_entry_common(box, r) + if r.remaining() >= 16: + r.skip(16) + if r.remaining() >= 4: + box.fields["width"] = r.u16(); box.fields["height"] = r.u16() + if r.remaining() >= 14: + box.fields["horiz_resolution"] = r.fixed16_16(); box.fields["vert_resolution"] = r.fixed16_16(); r.skip(4); box.fields["frame_count"] = r.u16() + if r.remaining() >= 32: + compressor = r.read(32) + box.fields["compressor_name"] = decode_text(compressor[1:1 + min(compressor[0], 31)]) if compressor else "" + if r.remaining() >= 4: + box.fields["depth"] = r.u16(); box.fields["pre_defined_3"] = r.i16() + consumed = r.tell() + child_start = box.payload_start + consumed + if child_start < box.end: + box.children = self._parse_children(child_start, box.end, box.type) + + def decode_mp4a(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r) + def decode_enca(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r) + def decode_alac(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r) + def decode_ac_3(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r) + def decode_ec_3(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r) + + def _decode_audio_sample_entry(self, box: Box, r: Reader) -> None: + self._decode_sample_entry_common(box, r) + if r.remaining() >= 8: + r.skip(8) + if r.remaining() >= 8: + box.fields["channel_count"] = r.u16(); box.fields["sample_size"] = r.u16(); r.skip(4) + if r.remaining() >= 4: + rate = r.u32() + box.fields["sample_rate"] = rate >> 16 + consumed = r.tell() + child_start = box.payload_start + consumed + if child_start < box.end: + box.children = self._parse_children(child_start, box.end, box.type) + + def decode_esds(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() > 0: + box.fields["descriptor_data"] = hex_preview(r.read(r.remaining())) + + def decode_avcC(self, box: Box, r: Reader) -> None: + if r.remaining() >= 7: + box.fields["configuration_version"] = r.u8(); box.fields["avc_profile_indication"] = r.u8(); box.fields["profile_compatibility"] = r.u8(); box.fields["avc_level_indication"] = r.u8() + length_size_minus_one = r.u8() & 3 + box.fields["nal_unit_length_size"] = length_size_minus_one + 1 + sps_count = r.u8() & 31 + sps = [] + for _ in range(sps_count): + if r.remaining() < 2: break + n = r.u16(); sps.append(r.read(min(n, r.remaining())).hex()) + pps_count = r.u8() if r.remaining() >= 1 else 0 + pps = [] + for _ in range(pps_count): + if r.remaining() < 2: break + n = r.u16(); pps.append(r.read(min(n, r.remaining())).hex()) + box.fields["sequence_parameter_sets"] = sps + box.fields["picture_parameter_sets"] = pps + + def decode_hvcC(self, box: Box, r: Reader) -> None: + if r.remaining() >= 23: + box.fields["configuration_version"] = r.u8() + byte = r.u8(); box.fields["general_profile_space"] = (byte >> 6) & 3; box.fields["general_tier_flag"] = (byte >> 5) & 1; box.fields["general_profile_idc"] = byte & 31 + box.fields["general_profile_compatibility_flags"] = r.u32() + box.fields["general_constraint_indicator_flags"] = r.read(6).hex() + box.fields["general_level_idc"] = r.u8() + r.skip(12) + arrays = r.u8() if r.remaining() else 0 + box.fields["num_arrays"] = arrays + if self.verbosity >= 2: + parsed = [] + for _ in range(arrays): + if r.remaining() < 3: break + hdr = r.u8(); count = r.u16(); units = [] + for _ in range(count): + if r.remaining() < 2: break + n = r.u16(); units.append(r.read(min(n, r.remaining())).hex()) + parsed.append({"nal_unit_type": hdr & 0x3F, "nal_units": units}) + box.fields["arrays"] = parsed + + def decode_dvcC(self, box: Box, r: Reader) -> None: self._decode_dovi(box, r) + def decode_dvvC(self, box: Box, r: Reader) -> None: self._decode_dovi(box, r) + + def _decode_dovi(self, box: Box, r: Reader) -> None: + if r.remaining() >= 4: + box.fields["dv_version_major"] = r.u8(); box.fields["dv_version_minor"] = r.u8() + value = r.u16() + box.fields["dv_profile"] = (value >> 9) & 0x7F + box.fields["dv_level"] = (value >> 3) & 0x3F + box.fields["rpu_present_flag"] = (value >> 2) & 1 + box.fields["el_present_flag"] = (value >> 1) & 1 + box.fields["bl_present_flag"] = value & 1 + + def decode_pasp(self, box: Box, r: Reader) -> None: + if r.remaining() >= 8: + box.fields["h_spacing"] = r.u32(); box.fields["v_spacing"] = r.u32() + + def decode_clli(self, box: Box, r: Reader) -> None: + if r.remaining() >= 4: + box.fields["max_content_light_level"] = r.u16(); box.fields["max_pic_average_light_level"] = r.u16() + + def decode_mdcv(self, box: Box, r: Reader) -> None: + if r.remaining() >= 24: + box.fields["display_primaries"] = [[r.u16(), r.u16()] for _ in range(3)] + box.fields["white_point"] = [r.u16(), r.u16()] + box.fields["max_display_mastering_luminance"] = r.u32() + box.fields["min_display_mastering_luminance"] = r.u32() + + def decode_btrt(self, box: Box, r: Reader) -> None: + if r.remaining() >= 12: + box.fields["buffer_size_db"] = r.u32(); box.fields["max_bitrate"] = r.u32(); box.fields["avg_bitrate"] = r.u32() + + def decode_colr(self, box: Box, r: Reader) -> None: + if r.remaining() >= 4: + colour_type = fourcc(r.read(4)); box.fields["colour_type"] = colour_type + if colour_type in {"nclx", "nclc"} and r.remaining() >= 6: + box.fields["colour_primaries"] = r.u16(); box.fields["transfer_characteristics"] = r.u16(); box.fields["matrix_coefficients"] = r.u16() + if colour_type == "nclx" and r.remaining() >= 1: + box.fields["full_range_flag"] = (r.u8() >> 7) & 1 + elif r.remaining() > 0: + box.fields["payload"] = hex_preview(r.read(r.remaining())) + + def decode_stts(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 8: break + entries.append({"sample_count": r.u32(), "sample_delta": r.u32()}) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + self._register("stts", entries) + + def decode_ctts(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 8: break + offset = r.i32() if box.version == 1 else r.u32() + entries.append({"sample_count": r.u32(), "sample_offset": offset}) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + + def decode_stsc(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 12: break + entries.append({"first_chunk": r.u32(), "samples_per_chunk": r.u32(), "sample_description_index": r.u32()}) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + self._register("stsc", entries) + + def decode_stsz(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + sample_size = r.u32() if r.remaining() >= 4 else 0 + sample_count = r.u32() if r.remaining() >= 4 else 0 + sizes = [] + if sample_size == 0: + for _ in range(sample_count): + if r.remaining() < 4: break + sizes.append(r.u32()) + else: + sizes = [sample_size] * sample_count + box.fields["sample_size"] = sample_size; box.fields["sample_count"] = sample_count + if self.verbosity >= 1: box.fields["entries"] = sizes + self._register("stsz", sizes) + + def decode_stz2(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 4: r.skip(3); field_size = r.u8() + else: field_size = 0 + sample_count = r.u32() if r.remaining() >= 4 else 0 + sizes = [] + if field_size == 4: + for _ in range((sample_count + 1) // 2): + if r.remaining() < 1: break + b = r.u8(); sizes.extend([(b >> 4) & 15, b & 15]) + sizes = sizes[:sample_count] + elif field_size == 8: + sizes = [r.u8() for _ in range(min(sample_count, r.remaining()))] + elif field_size == 16: + for _ in range(sample_count): + if r.remaining() < 2: break + sizes.append(r.u16()) + box.fields["field_size"] = field_size; box.fields["sample_count"] = sample_count + if self.verbosity >= 1: box.fields["entries"] = sizes + self._register("stsz", sizes) + + def decode_stco(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 4: break + entries.append(r.u32()) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + self._register("chunk_offsets", entries) + + def decode_co64(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 8: break + entries.append(r.u64()) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + self._register("chunk_offsets", entries) + + def decode_stss(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if r.remaining() < 4: break + entries.append(r.u32()) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + + def decode_sdtp(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if self.verbosity >= 2: + box.fields["sample_dependency_table"] = list(r.read(r.remaining())) + else: + box.fields["entry_count"] = r.remaining() + + def decode_elst(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = [] + for _ in range(count): + if box.version == 1: + if r.remaining() < 20: break + duration, media_time = r.u64(), r.i64() + else: + if r.remaining() < 12: break + duration, media_time = r.u32(), r.i32() + entries.append({"segment_duration": duration, "media_time": media_time, "media_rate": r.fixed16_16()}) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["entries"] = entries + + def decode_url(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() > 0: + box.fields["location"] = r.cstring() + + def decode_urn(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + box.fields["name"] = r.cstring() if r.remaining() else "" + box.fields["location"] = r.cstring() if r.remaining() else "" + + def decode_dref(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + box.fields["entry_count"] = r.u32() if r.remaining() >= 4 else 0 + + def decode_mfhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 4: box.fields["sequence_number"] = r.u32() + + def decode_tfhd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + track_id = r.u32() if r.remaining() >= 4 else 0 + self.current_track_id = track_id + flags = box.flags or 0 + box.fields["track_id"] = track_id + if flags & 0x000001 and r.remaining() >= 8: box.fields["base_data_offset"] = r.u64() + if flags & 0x000002 and r.remaining() >= 4: box.fields["sample_description_index"] = r.u32() + if flags & 0x000008 and r.remaining() >= 4: box.fields["default_sample_duration"] = r.u32() + if flags & 0x000010 and r.remaining() >= 4: box.fields["default_sample_size"] = r.u32() + if flags & 0x000020 and r.remaining() >= 4: box.fields["default_sample_flags"] = r.u32() + + def decode_tfdt(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if box.version == 1 and r.remaining() >= 8: box.fields["base_media_decode_time"] = r.u64() + elif r.remaining() >= 4: box.fields["base_media_decode_time"] = r.u32() + + def decode_trun(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + flags = box.flags or 0 + sample_count = r.u32() if r.remaining() >= 4 else 0 + box.fields["sample_count"] = sample_count + if flags & 0x000001 and r.remaining() >= 4: box.fields["data_offset"] = r.i32() + if flags & 0x000004 and r.remaining() >= 4: box.fields["first_sample_flags"] = r.u32() + entries = [] + for _ in range(sample_count): + entry: Dict[str, Any] = {} + if flags & 0x000100 and r.remaining() >= 4: entry["sample_duration"] = r.u32() + if flags & 0x000200 and r.remaining() >= 4: entry["sample_size"] = r.u32() + if flags & 0x000400 and r.remaining() >= 4: entry["sample_flags"] = r.u32() + if flags & 0x000800 and r.remaining() >= 4: entry["sample_composition_time_offset"] = r.i32() if box.version == 1 else r.u32() + entries.append(entry) + if self.verbosity >= 1: box.fields["entries"] = entries + + def decode_trex(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 20: + box.fields.update({"track_id": r.u32(), "default_sample_description_index": r.u32(), "default_sample_duration": r.u32(), "default_sample_size": r.u32(), "default_sample_flags": r.u32()}) + + def decode_mehd(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if box.version == 1 and r.remaining() >= 8: box.fields["fragment_duration"] = r.u64() + elif r.remaining() >= 4: box.fields["fragment_duration"] = r.u32() + + def decode_sidx(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() < 8: return + box.fields["reference_id"] = r.u32(); box.fields["timescale"] = r.u32() + if box.version == 1: + if r.remaining() < 16: return + box.fields["earliest_presentation_time"] = r.u64(); box.fields["first_offset"] = r.u64() + else: + if r.remaining() < 8: return + box.fields["earliest_presentation_time"] = r.u32(); box.fields["first_offset"] = r.u32() + if r.remaining() >= 2: r.u16() + count = r.u16() if r.remaining() >= 2 else 0 + refs = [] + for _ in range(count): + if r.remaining() < 12: break + a = r.u32(); b = r.u32(); c = r.u32() + refs.append({"reference_type": (a >> 31) & 1, "referenced_size": a & 0x7fffffff, "subsegment_duration": b, "starts_with_sap": (c >> 31) & 1, "sap_type": (c >> 28) & 7, "sap_delta_time": c & 0x0fffffff}) + box.fields["reference_count"] = count + if self.verbosity >= 1: box.fields["references"] = refs + + def decode_pssh(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() < 20: return + system_id = r.read(16).hex() + box.fields["system_id"] = system_id + box.fields["system_name"] = PROTECTION_SYSTEM_IDS.get(system_id, "") + if box.version and box.version > 0 and r.remaining() >= 4: + kid_count = r.u32(); kids = [] + for _ in range(kid_count): + if r.remaining() < 16: break + kids.append(r.read(16).hex()) + box.fields["kid_count"] = kid_count; box.fields["kids"] = kids + if r.remaining() >= 4: + data_size = r.u32(); data = r.read(min(data_size, r.remaining())) + box.fields["data_size"] = data_size + if self.verbosity >= 1: + box.fields["data"] = data.hex() + box.fields["data_base64"] = base64.b64encode(data).decode("ascii") + + def decode_tenc(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 2: + box.fields["reserved"] = r.u8() + second = r.u8() + if box.version and box.version > 0: + box.fields["default_crypt_byte_block"] = (second >> 4) & 0x0F + box.fields["default_skip_byte_block"] = second & 0x0F + else: + box.fields["reserved_2"] = second + if r.remaining() >= 1: box.fields["default_is_protected"] = r.u8() + if r.remaining() >= 1: box.fields["default_per_sample_iv_size"] = r.u8() + if r.remaining() >= 16: box.fields["default_kid"] = r.read(16).hex() + if box.fields.get("default_is_protected") == 1 and box.fields.get("default_per_sample_iv_size") == 0 and r.remaining() >= 1: + const_iv_size = r.u8(); box.fields["default_constant_iv_size"] = const_iv_size + if r.remaining() >= const_iv_size: box.fields["default_constant_iv"] = r.read(const_iv_size).hex() + + def decode_frma(self, box: Box, r: Reader) -> None: + if r.remaining() >= 4: box.fields["data_format"] = fourcc(r.read(4)) + + def decode_schm(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 8: + box.fields["scheme_type"] = fourcc(r.read(4)); box.fields["scheme_version"] = r.u32() + if (box.flags or 0) & 1 and r.remaining() > 0: + box.fields["scheme_uri"] = r.cstring() + + def decode_saiz(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if (box.flags or 0) & 1 and r.remaining() >= 8: + box.fields["aux_info_type"] = fourcc(r.read(4)); box.fields["aux_info_type_parameter"] = r.u32() + if r.remaining() >= 5: + default_size = r.u8(); sample_count = r.u32() + box.fields["default_sample_info_size"] = default_size; box.fields["sample_count"] = sample_count + if default_size == 0 and self.verbosity >= 1: + box.fields["sample_info_sizes"] = list(r.read(min(sample_count, r.remaining()))) + + def decode_saio(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if (box.flags or 0) & 1 and r.remaining() >= 8: + box.fields["aux_info_type"] = fourcc(r.read(4)); box.fields["aux_info_type_parameter"] = r.u32() + count = r.u32() if r.remaining() >= 4 else 0; offsets = [] + for _ in range(count): + need = 8 if box.version == 1 else 4 + if r.remaining() < need: break + offsets.append(r.u64() if box.version == 1 else r.u32()) + box.fields["entry_count"] = count + if self.verbosity >= 1: box.fields["offsets"] = offsets + + def decode_senc(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + count = r.u32() if r.remaining() >= 4 else 0 + box.fields["sample_count"] = count + if self.verbosity >= 2: + box.fields["raw"] = hex_preview(r.read(r.remaining()), 256) + + def decode_prft(self, box: Box, r: Reader) -> None: + self._read_full_box(box, r) + if r.remaining() >= 12: + box.fields["reference_track_id"] = r.u32(); box.fields["ntp_timestamp"] = r.u64() + if box.version == 1 and r.remaining() >= 8: box.fields["media_time"] = r.i64() + elif r.remaining() >= 4: box.fields["media_time"] = r.i32() + + def decode_uuid(self, box: Box, r: Reader) -> None: + if self.verbosity >= 2 and r.remaining() > 0: + box.fields["payload"] = hex_preview(r.read(r.remaining()), 256) + + def decode_data(self, box: Box, r: Reader) -> None: + if r.remaining() >= 8: + data_type = r.u32(); locale = r.u32(); payload = r.read(r.remaining()) + box.fields["data_type"] = data_type; box.fields["locale"] = locale + if data_type in {1, 2}: + box.fields["value"] = decode_text(payload) + elif data_type in {13, 14, 27}: + box.fields["value"] = payload.hex() + else: + box.fields["value"] = hex_preview(payload) + + def decode_data_container(self, box: Box, r: Reader) -> None: + if box.payload_size >= 8: + box.children = self._parse_children(box.payload_start, box.end, box.type) + + def decode_mdat(self, box: Box, r: Reader) -> None: + box.fields["data_size"] = box.payload_size + if self.verbosity >= 3 and box.payload_size: + box.fields["data_preview"] = hex_preview(r.read(min(box.payload_size, MAX_PREVIEW_BYTES))) + + def decode_free(self, box: Box, r: Reader) -> None: + box.fields["data_size"] = box.payload_size + + decode_skip = decode_free + +class TextPrinter: + def __init__(self, verbosity: int = 0) -> None: + self.verbosity = verbosity + + def print(self, boxes: List[Box]) -> None: + for box in boxes: + self._print_box(box, 0) + + def _print_box(self, box: Box, depth: int) -> None: + indent = " " * depth + extras = [] + if box.uuid: extras.append("uuid=%s" % box.uuid) + if box.version is not None: extras.append("version=%d" % box.version) + if box.flags is not None: extras.append("flags=0x%06x" % box.flags) + detail = " " + " ".join(extras) if extras else "" + print(f"{indent}[{box.type}] size={box.size} offset={box.start}{detail}") + if box.error: + print(f"{indent} error = {box.error}") + for key, value in box.fields.items(): + if self.verbosity == 0 and key in {"entries", "data", "data_base64", "payload", "sample_info_sizes", "offsets", "raw"}: + continue + self._print_field(key, value, depth + 1) + for child in box.children: + self._print_box(child, depth + 1) + + def _print_field(self, key: str, value: Any, depth: int) -> None: + indent = " " * depth + if isinstance(value, list) and len(value) > 16 and self.verbosity < 2: + print(f"{indent}{key} = [{len(value)} entries]") + return + text = json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value) + if len(text) > 240 and self.verbosity < 3: + text = text[:240] + "..." + print(f"{indent}{key} = {text}") + +class TrackDumper: + def __init__(self, data: bytes, parser: Mp4Parser) -> None: + self.data = data + self.parser = parser + + def dump_track(self, input_filename: str, track_id: int) -> str: + ctx = self.parser.track_context.get(track_id) + if not ctx: + raise Mp4DumpError("track not found: %d" % track_id) + sizes = ctx.get("stsz") or [] + offsets = ctx.get("chunk_offsets") or [] + stsc = ctx.get("stsc") or [] + if not sizes or not offsets or not stsc: + raise Mp4DumpError("track %d cannot be dumped because sample tables are incomplete or fragmented-only" % track_id) + sample_offsets = self._build_sample_offsets(sizes, offsets, stsc) + if len(sample_offsets) != len(sizes): + raise Mp4DumpError("track %d sample table mismatch" % track_id) + output_name = f"{input_filename}.{track_id}" + with open(output_name, "wb") as out: + for off, size in zip(sample_offsets, sizes): + if off < 0 or off + size > len(self.data): + raise Mp4DumpError("track %d sample offset outside file" % track_id) + out.write(struct.pack(">I", size)) + out.write(self.data[off:off + size]) + return output_name + + def _build_sample_offsets(self, sizes: List[int], chunk_offsets: List[int], stsc: List[Dict[str, int]]) -> List[int]: + sample_offsets: List[int] = [] + sample_index = 0 + entries = sorted(stsc, key=lambda e: e.get("first_chunk", 0)) + for entry_index, entry in enumerate(entries): + first_chunk = entry.get("first_chunk", 1) + next_first = entries[entry_index + 1].get("first_chunk", len(chunk_offsets) + 1) if entry_index + 1 < len(entries) else len(chunk_offsets) + 1 + samples_per_chunk = entry.get("samples_per_chunk", 0) + for chunk_number in range(first_chunk, next_first): + if chunk_number < 1 or chunk_number > len(chunk_offsets): + continue + offset = chunk_offsets[chunk_number - 1] + for _ in range(samples_per_chunk): + if sample_index >= len(sizes): + break + sample_offsets.append(offset) + offset += sizes[sample_index] + sample_index += 1 + return sample_offsets + +def parse_track_option(value: str) -> Tuple[int, Optional[str]]: + if ":" in value: + tid, key = value.split(":", 1) + else: + tid, key = value, None + try: + track_id = int(tid, 10) + except ValueError as exc: + raise argparse.ArgumentTypeError("invalid track id") from exc + if key: + clean = key.replace("-", "").replace(" ", "") + if len(clean) != 32: + raise argparse.ArgumentTypeError("track keys must be 128-bit hex values") + try: + binascii.unhexlify(clean) + except binascii.Error as exc: + raise argparse.ArgumentTypeError("invalid track key hex") from exc + return track_id, key + +def load_file(path: str) -> bytes: + with open(path, "rb") as f: + return f.read() + +PLAYREADY_SYSTEM_ID = "9a04f07998404286ab92e65be0885f95" +WIDEVINE_SYSTEM_ID = "edef8ba979d64acea3c827dcd51d21ed" + +def normalize_system_id(system_id: str) -> str: + return system_id.lower().replace("-", "").strip() + + +def walk_boxes(boxes: List[Box]) -> Iterable[Box]: + for box in boxes: + yield box + if box.children: + yield from walk_boxes(box.children) + + +def parse_mp4_boxes(input_file: str, verbosity: int = 1) -> List[Box]: + if not os.path.isfile(input_file): + raise Mp4DumpError("input file does not exist: %s" % input_file) + data = load_file(input_file) + parser = Mp4Parser(data, verbosity=verbosity) + return parser.parse() + + +def extract_pssh_boxes_from_mp4(input_file: str, system_id: Optional[str] = None) -> List[Dict[str, Any]]: + wanted = normalize_system_id(system_id) if system_id else None + if not os.path.isfile(input_file): + raise Mp4DumpError("input file does not exist: %s" % input_file) + + data = load_file(input_file) + parser = Mp4Parser(data, verbosity=1) + boxes = parser.parse() + results: List[Dict[str, Any]] = [] + + for box in walk_boxes(boxes): + if box.type != "pssh": + continue + + found_system_id = str(box.fields.get("system_id", "")).lower() + if wanted and found_system_id != wanted: + continue + + full_box = data[box.start:box.end] + item: Dict[str, Any] = { + "system_id": found_system_id, + "system_name": box.fields.get("system_name", ""), + "offset": box.start, + "size": box.size, + "pssh_base64": base64.b64encode(full_box).decode("ascii"), + "pssh_hex": full_box.hex(), + "data_size": box.fields.get("data_size"), + "data_base64": box.fields.get("data_base64"), + "data_hex": box.fields.get("data"), + } + if "kids" in box.fields: + item["kids"] = box.fields.get("kids") + if "kid_count" in box.fields: + item["kid_count"] = box.fields.get("kid_count") + results.append(item) + + return results + + +def extract_playready_pssh_from_mp4(input_file: str) -> Optional[str]: + matches = extract_pssh_boxes_from_mp4(input_file, PLAYREADY_SYSTEM_ID) + if not matches: + return None + value = matches[0].get("pssh_base64") + return str(value) if value is not None else None + + +def extract_playready_pssh_payload_from_mp4(input_file: str) -> Optional[str]: + matches = extract_pssh_boxes_from_mp4(input_file, PLAYREADY_SYSTEM_ID) + if not matches: + return None + value = matches[0].get("data_base64") + return str(value) if value is not None else None + + +def extract_playready_pssh_from_init(input_file: str) -> Optional[str]: + return extract_playready_pssh_from_mp4(input_file) + + +def extract_playready_pssh_payload_from_init(input_file: str) -> Optional[str]: + return extract_playready_pssh_payload_from_mp4(input_file) + +def build_arg_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="pymp4dump.py", + description="Python MP4/QuickTime atom dumper inspired by Bento4 mp4dump.", + formatter_class=argparse.RawTextHelpFormatter, + ) + p.add_argument("input", help="input MP4/QuickTime/ISO BMFF file") + p.add_argument("--verbosity", type=int, default=0, choices=[0, 1, 2, 3], help="sets the detail level between 0 and 3") + p.add_argument("--format", choices=["text", "json"], default="text", help="output format: text or json") + p.add_argument("--track", action="append", default=[], metavar="TRACK_ID[:KEY]", help="dump a track to .; KEY is accepted for Bento4 CLI compatibility but decryption is not performed") + p.add_argument("--no-banner", action="store_true", help="do not print the banner in text mode") + p.add_argument("--extract-playready-pssh", action="store_true", help="print the first complete PlayReady PSSH box as base64, matching pypr") + p.add_argument("--extract-playready-payload", action="store_true", help="print only the first inner PlayReady PSSH data payload as base64") + return p + +def main(argv: Optional[List[str]] = None) -> int: + args = build_arg_parser().parse_args(argv) + if not os.path.isfile(args.input): + print("ERROR: input file does not exist: %s" % args.input, file=sys.stderr) + return 1 + try: + track_specs = [parse_track_option(x) for x in args.track] + if args.extract_playready_pssh: + pssh = extract_playready_pssh_from_mp4(args.input) + if pssh is None: + print("ERROR: no PlayReady PSSH found", file=sys.stderr) + return 1 + print(pssh) + return 0 + if args.extract_playready_payload: + payload = extract_playready_pssh_payload_from_mp4(args.input) + if payload is None: + print("ERROR: no PlayReady PSSH payload found", file=sys.stderr) + return 1 + print(payload) + return 0 + data = load_file(args.input) + parser = Mp4Parser(data, verbosity=args.verbosity) + boxes = parser.parse() + if args.format == "json": + print(json.dumps([box.to_json() for box in boxes], indent=2, ensure_ascii=False)) + else: + if not args.no_banner: + print(BANNER) + print() + TextPrinter(args.verbosity).print(boxes) + if track_specs: + dumper = TrackDumper(data, parser) + for track_id, key in track_specs: + if key: + print("WARNING: decryption keys are accepted for CLI compatibility, but this Python implementation writes stored sample bytes only.", file=sys.stderr) + output = dumper.dump_track(args.input, track_id) + print("Dumped track %d to %s" % (track_id, output), file=sys.stderr) + return 0 + except (OSError, Mp4DumpError, argparse.ArgumentTypeError) as exc: + print("ERROR: %s" % exc, file=sys.stderr) + return 1 + except KeyboardInterrupt: + print("Interrupted", file=sys.stderr) + return 130 + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file