forked from Manuel/meeting-assistant
182 lines
9.3 KiB
Python
182 lines
9.3 KiB
Python
"""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"<?xml"):
|
|
raise RuntimeError("Unsupported UDIF metadata framing")
|
|
self.blocks = plist["resource-fork"]["blkx"]
|
|
self.physical_runs = {}
|
|
intervals = []
|
|
for block in self.blocks:
|
|
mish = block["Data"]
|
|
if len(mish) < 244 or mish[:4] != b"mish" or (len(mish) - 204) % 40 or u32(mish, 200) != (len(mish) - 204) // 40:
|
|
raise RuntimeError("Malformed UDIF block table")
|
|
crc_contract(mish, 64)
|
|
if u64(mish, 8) + u64(mish, 16) > 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"<data>([\sA-Za-z0-9+/=]*)</data>", 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))
|