Created
April 19, 2026 09:59
-
-
Save jernejsk/9ec858d1c4746062194d7806fc2d312d to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import configparser | |
| from collections import defaultdict | |
| import binascii | |
| import re | |
| import struct | |
| import sys | |
| import uuid | |
| total_sectors = 8 * 1024 * 1024 * 2 # assume 8G image | |
| SECTOR = 512 | |
| PRIMARY_HDR_LBA = 1 | |
| PRIMARY_ENT_LBA = 2 | |
| ENTRIES_SECTORS = 32 # 128 entries × 128 bytes / 512 | |
| MAX_ENTRIES = 128 | |
| ENTRY_SIZE = 128 | |
| FIRST_USABLE = PRIMARY_ENT_LBA + ENTRIES_SECTORS # LBA 34 — fixed, never changes | |
| signature = b'IMAGEWTY' | |
| header_ver = 0x300 | |
| header_size = 96 | |
| format_ver = 0x100234 | |
| part_magic = b'softw411' | |
| part_ver = 0x200 | |
| chunk_size = 1024 * 1024 | |
| sparse_magic = 0xED26FF3A | |
| gpt_file = ("12345678", "1234567890___GPT") | |
| gpt_desc = ("COMMON ", "SYS_CONFIG000000") | |
| script_file = ("12345678", "1234567890SCRIPT") | |
| dlinfo_file = ("12345678", "1234567890DLINFO") | |
| dlmain = "RFSFAT16" | |
| class PhoenixImage: | |
| def __init__(self): | |
| self.files = {} | |
| self.fd = None | |
| def parse(self, fn): | |
| if self.fd: | |
| self.fd.close() | |
| self.fd = open(fn, 'rb') | |
| # validate file size | |
| self.fd.seek(0, 2) | |
| size = self.fd.tell() | |
| self.fd.seek(0, 0) | |
| if size < header_size: | |
| print("File too short!") | |
| return False | |
| # validate common header part | |
| hdr = self.fd.read(32) | |
| sig, hdr_ver, hdr_size, fmt_ver, img_size = struct.unpack("<8sLL4xLL4x", hdr) | |
| if sig != signature: | |
| print("Not an PhoenixCard image! Is it encrypted?") | |
| return False | |
| if hdr_ver != header_ver: | |
| print(f"Unknown header version: {hex(hdr_ver)}") | |
| return False | |
| if hdr_size != header_size: | |
| print(f"Invalid header size! Should be {header_size}, currenty {hdr_size}") | |
| return False | |
| if fmt_ver != format_ver: | |
| print(f"Unknown format version: {hex(fmt_ver)}") | |
| return False | |
| if size < img_size: | |
| print("File is too short!") | |
| return False | |
| # read file count from header v3 part | |
| hdr = self.fd.read(64) | |
| count = struct.unpack('<28xL32x', hdr)[0] | |
| # header is aligned to 1024 | |
| self.fd.seek(1024, 0) | |
| self.files = {} | |
| for i in range(count): | |
| hdr = self.fd.read(32) | |
| fn_len, hdr_size, main, sub = struct.unpack("<LL8s16s", hdr) | |
| hdr = self.fd.read(24 + fn_len) | |
| length, offset = struct.unpack(f"<{fn_len + 12}xL4xL", hdr) | |
| self.files[(main.decode('ascii'), sub.decode('ascii'))] = (offset, length) | |
| self.fd.seek(hdr_size - 56 - fn_len, 1) | |
| return True | |
| def read_file(self, part_file): | |
| info = self.files[part_file] | |
| self.fd.seek(info[0], 0) | |
| return self.fd.read(info[1]) | |
| def part_exists(self, part_file): | |
| return part_file in self.files | |
| def __copy_chunk(self, count, of): | |
| while (count): | |
| to_read = count | |
| if count > chunk_size: | |
| to_read = chunk_size | |
| count -= to_read | |
| data = self.fd.read(to_read) | |
| of.write(data) | |
| def __copy_sparse(self, count, of): | |
| header = struct.unpack("<I4H4I", self.fd.read(28)) | |
| bsize = header[5] | |
| for i in range(header[7]): | |
| ctype, _, csize, _ = struct.unpack("<2H2I", self.fd.read(12)) | |
| if ctype == 0xcac1: | |
| self.__copy_chunk(csize * bsize, of) | |
| elif ctype == 0xcac2: | |
| data = self.fd.read(4) * (bsize // 4) | |
| for blk in range(csize): | |
| of.write(data) | |
| elif ctype == 0xcac3: | |
| of.seek(csize * bsize, 1) | |
| elif ctype == 0xcac4: | |
| self.fd.read(4) | |
| def copy_part(self, part_file, of, offset): | |
| info = self.files[part_file] | |
| self.fd.seek(info[0], 0) | |
| of.seek(offset, 0) | |
| magic = struct.unpack("<L", self.fd.read(4))[0] | |
| self.fd.seek(info[0], 0) | |
| if magic == sparse_magic: | |
| self.__copy_sparse(info[1], of) | |
| else: | |
| self.__copy_chunk(info[1], of) | |
| def __del__(self): | |
| if self.fd: | |
| self.fd.close() | |
| def _strip_comment(line: str) -> str: | |
| """Remove `;` comments that aren't inside double-quoted strings.""" | |
| in_quotes = False | |
| for i, ch in enumerate(line): | |
| if ch == '"': | |
| in_quotes = not in_quotes | |
| elif ch == ";" and not in_quotes: | |
| return line[:i] | |
| return line | |
| def _coerce(value: str): | |
| """Try to convert to int (hex or decimal), fall back to str.""" | |
| try: | |
| return int(value, 0) # handles 0x… and plain integers | |
| except (ValueError, TypeError): | |
| return value | |
| def parse_fex(text: str) -> dict: | |
| """ | |
| Parse a .fex file into a dict where: | |
| - Unique sections → dict of key/value pairs | |
| - Duplicate sections → list of dicts | |
| """ | |
| # Track sections; use a list to preserve order and allow duplicates | |
| sections: list[tuple[str, dict]] = [] # [(section_name, {key: value}), ...] | |
| current_name: str | None = None | |
| current_data: dict = {} | |
| for raw_line in text.splitlines(): | |
| # Strip inline comments (`;` outside quotes) and whitespace | |
| line = _strip_comment(raw_line).strip() | |
| if not line: | |
| continue | |
| # Section header | |
| if line.startswith("[") and line.endswith("]"): | |
| if current_name is not None: | |
| sections.append((current_name, current_data)) | |
| current_name = line[1:-1].strip() | |
| current_data = {} | |
| continue | |
| # Key = value | |
| if "=" in line and current_name is not None: | |
| key, _, value = line.partition("=") | |
| key = key.strip() | |
| value = value.strip() | |
| # Remove surrounding quotes if present | |
| if len(value) >= 2 and value[0] == '"' and value[-1] == '"': | |
| value = value[1:-1] | |
| current_data[key] = _coerce(value) | |
| # Don't forget the last section | |
| if current_name is not None: | |
| sections.append((current_name, current_data)) | |
| # Collapse into final dict — duplicates become lists | |
| counts: dict[str, int] = defaultdict(int) | |
| for name, _ in sections: | |
| counts[name] += 1 | |
| result: dict = {} | |
| accumulators: dict[str, list] = {} | |
| for name, data in sections: | |
| if counts[name] == 1: | |
| result[name] = data | |
| else: | |
| if name not in accumulators: | |
| accumulators[name] = [] | |
| result[name] = accumulators[name] | |
| accumulators[name].append(data) | |
| return result | |
| def parse_partitions(image): | |
| data = image.read_file(dlinfo_file) | |
| hdr = data[:32] | |
| ver, magic, count = struct.unpack("<4xL8sL12x", hdr) | |
| if magic != part_magic: | |
| print(f"Unknown partitions magic {magic.decode('ascii')}") | |
| return [] | |
| if ver != part_ver: | |
| print(f"Unknown partitions version: {hex(ver)}") | |
| return [] | |
| parts = [] | |
| for i in range(count): | |
| offset = 32 + i * 72 | |
| info = data[offset : offset + 72] | |
| name, dlname = struct.unpack("<16s16x16s24x", info) | |
| name = name.decode('ascii').strip('\x00') | |
| dlname = dlname.decode('ascii').strip('\x00') | |
| parts.append((name, dlname)) | |
| return parts | |
| def parse_gpt_bin(image): | |
| data = image.read_file(gpt_file) | |
| hdr = data[0x200:0x25c] | |
| backup, start, count, part_size = struct.unpack("<32xQ32xQLL4x", hdr) | |
| partitions = {} | |
| offset = 512 * start | |
| for i in range(count): | |
| hdr = data[offset:offset+part_size] | |
| offset += part_size | |
| start, end, name = struct.unpack("<32x2Q8x72s", hdr) | |
| name = name.decode("utf-16").strip('\x00') | |
| partitions[name] = [i, start, end] | |
| if (end + 1 > backup): | |
| backup = end + 1 | |
| # this shouldn't be needed, but broken GPT exists | |
| if (start + 1 > backup): | |
| backup = start + 1 | |
| return backup, partitions | |
| def parse_gpt_conf(image): | |
| data = image.read_file(gpt_desc) | |
| config = parse_fex(data.decode("utf-8").strip('\x00')) | |
| gpt_cfg = config["gpt"] | |
| align = gpt_cfg["align"] | |
| cursor = gpt_cfg["start"] | |
| partitions = {} | |
| for i, p in enumerate(config.get("partition", [])): | |
| part_align = p.get("align", align) | |
| start = ((cursor + part_align - 1) // part_align) * part_align | |
| size = p["size"] if p["size"] > 0 else (total_sectors - 34 - start) | |
| end = start + size - 1 | |
| cursor = start + size | |
| type_guid = uuid.UUID(p.get("type") or "00000000-0000-0000-0000-000000000000") | |
| uniq_guid = uuid.UUID(p.get("uniqueguid") or str(uuid.uuid4())) | |
| attrs = (p.get("attrhi", 0) << 32) | p.get("attrlo", 0) | |
| name_utf16 = p["name"].encode("utf-16-le").ljust(72, b"\x00")[:72] | |
| partitions[p['name']] = [i, start, end, type_guid, uniq_guid, attrs, name_utf16] | |
| return cursor, partitions | |
| def parse_gpt(image, first_lba): | |
| if image.part_exists(gpt_file): | |
| return parse_gpt_bin(image) | |
| return parse_gpt_conf(image) | |
| def write_gpt_bin(image, size, partitions, of): | |
| data = bytearray(image.read_file(gpt_file)) | |
| # fix partition ends | |
| for part in partitions.values(): | |
| if (part[1] > part[2]): | |
| offset = 0x428 + part[0] * 128 | |
| data[offset:offset+8] = struct.pack("<Q", part[1]) | |
| # update partitions crc | |
| crc = binascii.crc32(data[0x400:0x400 + len(partitions) * 128]) & 0xFFFFFFFF | |
| data[0x258:0x25c] = struct.pack("<L", crc) | |
| backup = struct.unpack("<Q", data[0x220:0x228])[0] | |
| if (backup < size): | |
| # backup gpt lba | |
| data[0x220:0x228] = struct.pack("<Q", size) | |
| # last usable lba | |
| data[0x230:0x238] = struct.pack("<Q", size - 1) | |
| # update header crc | |
| data[0x210:0x214] = b'\x00\x00\x00\x00' | |
| crc = binascii.crc32(data[0x200:0x25c]) & 0xFFFFFFFF | |
| data[0x210:0x214] = struct.pack("<L", crc) | |
| of.seek(0, 0) | |
| of.write(data) | |
| def write_gpt_conf(size, partitions, of): | |
| raw_entries: list[bytes] = [] | |
| for part in partitions.values(): | |
| raw_entries.append(struct.pack( | |
| "<16s16sQQQ72s", | |
| part[3].bytes_le, | |
| part[4].bytes_le, | |
| part[1], part[2], part[5], | |
| part[6], | |
| )) | |
| entry_data = b"".join(raw_entries).ljust(MAX_ENTRIES * ENTRY_SIZE, b"\x00") | |
| entries_crc = binascii.crc32(entry_data) & 0xFFFFFFFF | |
| disk_guid = uuid.uuid4() | |
| def make_header(my_lba: int, alt_lba: int, entries_start_lba: int) -> bytes: | |
| hdr = struct.pack( | |
| "<8sIIIIQQQQ16sQIII", | |
| b"EFI PART", | |
| 0x00010000, | |
| 92, | |
| 0, | |
| 0, | |
| my_lba, | |
| alt_lba, | |
| FIRST_USABLE, | |
| size, | |
| disk_guid.bytes_le, | |
| entries_start_lba, | |
| MAX_ENTRIES, | |
| ENTRY_SIZE, | |
| entries_crc, | |
| ) | |
| # CRC covers exactly the first 92 bytes, with the CRC field itself zeroed | |
| crc = binascii.crc32(hdr) & 0xFFFFFFFF | |
| return hdr[:16] + struct.pack("<I", crc) + hdr[20:] | |
| backup_entries_lba = size + 1 | |
| backup_gpt_lba = backup_entries_lba + ENTRIES_SECTORS | |
| primary_header = make_header( | |
| my_lba = PRIMARY_HDR_LBA, | |
| alt_lba = backup_gpt_lba, | |
| entries_start_lba = PRIMARY_ENT_LBA, | |
| ) | |
| backup_header = make_header( | |
| my_lba = backup_gpt_lba, | |
| alt_lba = PRIMARY_HDR_LBA, | |
| entries_start_lba = backup_entries_lba, | |
| ) | |
| # ── Protective MBR ─────────────────────────────────────────────────────── | |
| pmbr = bytearray(SECTOR) | |
| # Single partition entry at offset 446: type 0xEE covering the whole disk | |
| pmbr[446] = 0x00 # not bootable | |
| pmbr[447:450] = b"\x00\x02\x00" # CHS start (head=0, sec=2, cyl=0) | |
| pmbr[450] = 0xEE # partition type: GPT protective | |
| pmbr[451:454] = b"\xFF\xFF\xFF" # CHS end (maxed out) | |
| struct.pack_into("<I", pmbr, 454, 1) # LBA start = 1 | |
| struct.pack_into("<I", pmbr, 458, min(backup_gpt_lba, 0xFFFF_FFFF)) | |
| pmbr[510] = 0x55 | |
| pmbr[511] = 0xAA | |
| # ── Assemble blobs ─────────────────────────────────────────────────────── | |
| # Primary: sector 0 (MBR) + sector 1 (header) + sectors 2-33 (entries) | |
| primary_blob = ( | |
| bytes(pmbr) | |
| + primary_header.ljust(SECTOR, b"\x00") | |
| + entry_data # exactly 32 sectors | |
| ) | |
| assert len(primary_blob) == 34 * SECTOR | |
| # Backup: sectors (total-33)..(total-2) (entries) + sector (total-1) (header) | |
| backup_blob = ( | |
| entry_data # 32 sectors of entries first | |
| + backup_header.ljust(SECTOR, b"\x00") # then the backup header | |
| ) | |
| assert len(backup_blob) == 33 * SECTOR | |
| of.seek(0, 0) | |
| of.write(primary_blob) | |
| of.seek(backup_entries_lba * SECTOR, 0) | |
| of.write(backup_blob) | |
| def write_gpt(image, size, partitions, of): | |
| if image.part_exists(gpt_file): | |
| return write_gpt_bin(image, size, partitions, of) | |
| return write_gpt_conf(size, partitions, of) | |
| def parse_script(image): | |
| data = image.read_file(script_file) | |
| config = configparser.ConfigParser() | |
| config.read_string(data.decode("utf-8").strip('\x00')) | |
| boot = [] | |
| for section in config.sections(): | |
| if section.startswith('boot'): | |
| item = config[section] | |
| boot.append(((item['main'], item['sub']), int(item['start']))) | |
| return int(config['card_boot']['start']), boot | |
| def create_image(image, of): | |
| parts = parse_partitions(image) | |
| size, partitions = parse_gpt(image) | |
| #boot = parse_script(image) | |
| of.truncate(512 * (size + 1)) | |
| write_gpt(image, size, partitions, of) | |
| for name, dlname in parts: | |
| lba = partitions[name][1] | |
| image.copy_part((dlmain, dlname), of, lba * 512) | |
| #for item in boot: | |
| # image.copy_part(item[0], of, item[1] * 512) | |
| def convert_image(infn, outfn): | |
| image = PhoenixImage() | |
| if not image.parse(infn): | |
| print("Can't parse input image!") | |
| return | |
| with open(outfn, 'wb') as of: | |
| create_image(image, of) | |
| if __name__ == '__main__': | |
| if (len(sys.argv) != 3): | |
| print(f"Usage: {sys.argv[0]} <phoenix image> <output image>") | |
| sys.exit(0) | |
| convert_image(sys.argv[1], sys.argv[2]) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment