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