"""Checksum binding for the pinned Linux Recovery UDIF patcher, not a new CLI. Source semantics: planetbeing/libdmg-hfsplus dmg/dmglib.c and dmg/blkx.c. Only flattened, single-segment XML UDIF with CRC32 and raw/zlib data is accepted. The caller plans same-length chunk writes; XML formatting and all offsets remain. """ import base64 import json import plistlib import re import struct import zlib def u32(data, offset): return struct.unpack_from(">I", data, offset)[0] def u64(data, offset): return struct.unpack_from(">Q", data, offset)[0] def crc_contract(data, offset): if u32(data, offset) != 2 or u32(data, offset + 4) != 32 or any(data[offset + 12:offset + 136]): raise RuntimeError("Unsupported UDIF checksum type/size/padding") return u32(data, offset + 8) class ChecksumPlan: def __init__(self, image, koly, plist, xml_offset, xml_length, size): self.image, self.koly, self.plist = image, koly, plist self.xml_offset, self.xml_length, self.size = xml_offset, xml_length, size self.data_offset, self.data_length = u64(koly, 24), u64(koly, 32) if (u32(koly, 4) != 4 or u32(koly, 8) != 512 or u32(koly, 12) != 1 or self.data_offset != 0 or u64(koly, 40) or u64(koly, 48) or u32(koly, 60) not in (0, 1) or self.data_length != xml_offset or xml_offset + xml_length > size - 512 or xml_length > 8 * 1024 * 1024): raise RuntimeError("Unsupported or out-of-bounds flattened UDIF layout") crc_contract(koly, 80) crc_contract(koly, 352) image.seek(xml_offset) self.xml = image.read(xml_length) if len(self.xml) != xml_length or not self.xml.lstrip().startswith(b" u64(koly, 492): raise RuntimeError("UDIF partition exceeds logical disk boundary") count = u32(mish, 200) if u32(mish, 204 + (count - 1) * 40) != 0xffffffff: raise RuntimeError("UDIF block table has no final terminator") for kind, offset, length, sectors in self.runs(mish): if kind in (2, 0x7ffffffe, 0xffffffff): if length: raise RuntimeError("Non-data UDIF run has stored bytes") continue if kind not in (1, 0x80000005) or not sectors or length <= 0 or sectors * 512 > 32 * 1024 * 1024: raise RuntimeError("Unsupported UDIF compression/run boundary") if offset < self.data_offset or offset + length > self.data_offset + self.data_length: raise RuntimeError("UDIF data run exceeds data-fork boundary") intervals.append((offset, offset + length)) self.physical_runs[offset] = length intervals.sort() if any(left[1] > right[0] for left, right in zip(intervals, intervals[1:])): raise RuntimeError("Overlapping UDIF physical data runs") def runs(self, mish): for entry in range(204, len(mish), 40): kind = u32(mish, entry) sector, sectors = u64(mish, entry + 8), u64(mish, entry + 16) if sector + sectors > u64(mish, 16): raise RuntimeError("UDIF run exceeds its partition boundary") offset = self.data_offset + u64(mish, 24) + u64(mish, entry + 24) yield kind, offset, u64(mish, entry + 32), sectors def read(self, offset, length): self.image.seek(offset) result = self.image.read(length) if len(result) != length: raise RuntimeError("Short UDIF checksum read") return result def logical_crcs(self, mish, replacements): original_crc = patched_crc = 0 for kind, offset, length, sectors in self.runs(mish): # IGNORE runs are excluded by the independently verified Apple 13 # baseline. Unknown ZERO/compression types are rejected above. if kind not in (1, 0x80000005): continue old = self.read(offset, length) new = replacements.get(offset, old) old_decoded = old if kind == 1 else zlib.decompress(old) new_decoded = new if kind == 1 else zlib.decompress(new) if len(old_decoded) != sectors * 512 or len(new_decoded) != sectors * 512 or len(old) != len(new): raise RuntimeError("UDIF checksum run changed physical/logical extent") original_crc = zlib.crc32(old_decoded, original_crc) patched_crc = zlib.crc32(new_decoded, patched_crc) return original_crc, patched_crc def data_crcs(self, replacements): old_crc = new_crc = 0 cursor = self.data_offset def same_until(stop): nonlocal cursor, old_crc, new_crc while cursor < stop: data = self.read(cursor, min(1024 * 1024, stop - cursor)) old_crc, new_crc = zlib.crc32(data, old_crc), zlib.crc32(data, new_crc) cursor += len(data) for offset, new in sorted(replacements.items()): same_until(offset) old = self.read(offset, len(new)) old_crc, new_crc = zlib.crc32(old, old_crc), zlib.crc32(new, new_crc) cursor += len(new) same_until(self.data_offset + self.data_length) return old_crc, new_crc def prepare(self, planned): replacements = dict(planned) if len(replacements) != len(planned): raise RuntimeError("Duplicate planned UDIF physical writes") if any(self.physical_runs.get(offset) != len(data) for offset, data in replacements.items()): raise RuntimeError("Planned UDIF write does not preserve an existing data-run boundary") old_master = bytearray() new_master = bytearray() changed = [] for block in self.blocks: mish = block["Data"] old_crc, new_crc = self.logical_crcs(mish, replacements) if old_crc != crc_contract(mish, 64): raise RuntimeError("Original UDIF logical CRC32 mismatch") old_master.extend(struct.pack(">I", old_crc)) new_master.extend(struct.pack(">I", new_crc)) if new_crc != old_crc: new_mish = bytearray(mish) struct.pack_into(">I", new_mish, 72, new_crc) changed.append((mish, bytes(new_mish))) old_data_crc, new_data_crc = self.data_crcs(replacements) if old_data_crc != crc_contract(self.koly, 80) or zlib.crc32(old_master) != crc_contract(self.koly, 352): raise RuntimeError("Original UDIF data-fork/master CRC32 mismatch") xml = self.xml for old_mish, new_mish in changed: matches = [match for match in re.finditer(rb"([\sA-Za-z0-9+/=]*)", xml) if base64.b64decode(match.group(1)) == old_mish] if len(matches) != 1: raise RuntimeError("UDIF block checksum XML identity is ambiguous") match = matches[0] encoded = iter(base64.b64encode(new_mish)) text = bytes(value if chr(value).isspace() else next(encoded) for value in match.group(1)) xml = xml[:match.start(1)] + text + xml[match.end(1):] if len(xml) != self.xml_length: raise RuntimeError("UDIF checksum update changed XML region length") new_plist = plistlib.loads(xml) expected = dict(self.plist) expected["resource-fork"] = dict(self.plist["resource-fork"]) expected["resource-fork"]["blkx"] = [dict(block, Data=dict(changed).get(block["Data"], block["Data"])) for block in self.blocks] if new_plist != expected: raise RuntimeError("UDIF checksum update changed unrelated metadata") koly = bytearray(self.koly) struct.pack_into(">I", koly, 88, new_data_crc) struct.pack_into(">I", koly, 360, zlib.crc32(new_master)) self.receipt = dict(originalDataCrc32=f"{old_data_crc:08x}", patchedDataCrc32=f"{new_data_crc:08x}", originalMasterCrc32=f"{zlib.crc32(old_master):08x}", patchedMasterCrc32=f"{zlib.crc32(new_master):08x}", changedBlockChecksums=len(changed), imageBytes=self.size, xmlOffset=self.xml_offset, xmlBytes=self.xml_length, physicalAndLogicalExtentsPreserved=True) return xml, bytes(koly) def verify(self): koly = self.read(self.size - 512, 512) xml = self.read(self.xml_offset, self.xml_length) verifier = ChecksumPlan(self.image, koly, plistlib.loads(xml), self.xml_offset, self.xml_length, self.size) verifier.prepare([]) self.image.seek(0, 2) if self.image.tell() != self.size: raise RuntimeError("Patched UDIF image length changed") print("[recovery-udif] " + json.dumps(dict(self.receipt, readBackChecksumsVerified=True), sort_keys=True))