firmware/kingst-la/sigrok-fwextract-kingst-la2016

   1 #!/usr/bin/python3
   2 ##
   3 ## This file is part of the sigrok-util project.
   4 ##
   5 ## Copyright (C) 2020 Florian Schmidt <schmidt_florian@gmx.de>
   6 ##
   7 ## This program is free software; you can redistribute it and/or modify
   8 ## it under the terms of the GNU General Public License as published by
   9 ## the Free Software Foundation; either version 3 of the License, or
  10 ## (at your option) any later version.
  11 ##
  12 ## This program is distributed in the hope that it will be useful,
  13 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 ## GNU General Public License for more details.
  16 ##
  17 ## You should have received a copy of the GNU General Public License
  18 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  19 ##
  20
  21 # This utility extracts FX2 MCU firmware and FPGA bitstream images from
  22 # the "KingstVIS" vendor software. The blobs are kept in Qt resources
  23 # sections. The script was tested with several v3.5 software versions.
  24
  25 import argparse
  26 import os
  27 import sys
  28 import re
  29 import struct
  30 import codecs
  31 import importlib.util
  32 import zlib
  33
  34 # Reuse the parseelf.py module from saleae-logic16.
  35 fwdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  36 parseelf_py = os.path.join(fwdir, "saleae-logic16", "parseelf.py")
  37 spec = importlib.util.spec_from_file_location("parseelf", parseelf_py)
  38 parseelf = importlib.util.module_from_spec(spec)
  39 spec.loader.exec_module(parseelf)
  40
  41 class qt_resources(object):
  42     def __init__(self, program):
  43         self._elf = parseelf.elf(program)
  44         self._elf_sections = {} # idx -> data
  45         self._read_resources()
  46
  47     def _get_elf_section(self, idx):
  48         s = self._elf_sections.get(idx)
  49         if s is None:
  50             shdr = self._elf.shdrs[idx]
  51             s = self._elf.read_section(shdr), shdr
  52             self._elf_sections[idx] = s
  53         return s
  54
  55     def _get_elf_sym_value(self, sname):
  56         sym = self._elf.symtab[sname]
  57         section, shdr = self._get_elf_section(sym["st_shndx"])
  58         addr = sym["st_value"] - shdr["sh_addr"]
  59         value = section[addr:addr + sym["st_size"]]
  60         if len(value) != sym["st_size"]:
  61             print("warning: symbol %s should be %d bytes, but in section is only %d bytes" % (
  62                 sname, sym["st_size"], len(value)))
  63         return value
  64
  65     # Qt resource stuff.
  66     def _get_resource_name(self, offset):
  67         length, i = struct.unpack(">HI", self._res_names[offset:offset + 2 + 4])
  68         offset += 2 + 4
  69         name = self._res_names[offset:offset + 2 * length].decode("utf-16be")
  70         return name
  71
  72     def _get_resource_data(self, offset):
  73         length = struct.unpack(">I", self._res_datas[offset:offset + 4])[0]
  74         offset += 4
  75         return self._res_datas[offset:offset + length]
  76
  77     def _read_resources(self):
  78         RCCFileInfo_Directory = 0x02
  79         def read_table():
  80             table = []
  81             offset = 0
  82             while offset < len(self._res_struct):
  83                 name_offset, flags = struct.unpack(">IH", self._res_struct[offset:offset+4+2])
  84                 offset += 6
  85                 name = self._get_resource_name(name_offset)
  86                 if flags & RCCFileInfo_Directory:
  87                     child_count, first_child_offset = struct.unpack(">II", self._res_struct[offset:offset + 4 + 4])
  88                     offset += 4 + 4
  89                     table.append((name, flags, child_count, first_child_offset))
  90                 else:
  91                     country, language, data_offset = struct.unpack(">HHI", self._res_struct[offset:offset + 2 + 2 + 4])
  92                     offset += 2 + 2 + 4
  93                     table.append((name, flags, country, language, data_offset))
  94             return table
  95         def read_dir_entries(table, which, parents=[]):
  96             name, flags = which[:2]
  97             if not flags & RCCFileInfo_Directory:
  98                 raise Exception("not a directory!")
  99             child_count, first_child = which[2:]
 100             for i in range(child_count):
 101                 child = table[first_child + i]
 102                 if child[1] & RCCFileInfo_Directory:
 103                     read_dir_entries(table, child, parents + [child[0]])
 104                 else:
 105                     country, language, data_offset = child[2:]
 106                     full_name = "/".join(parents + [child[0]])
 107                     self._resources[full_name] = data_offset
 108
 109         self._res_datas = self._get_elf_sym_value("_ZL16qt_resource_data")
 110         self._res_names = self._get_elf_sym_value("_ZL16qt_resource_name")
 111         self._res_struct = self._get_elf_sym_value("_ZL18qt_resource_struct")
 112
 113         self._resources = {} # res_fn -> res_offset
 114         table = read_table()
 115         read_dir_entries(table, table[0])
 116
 117     def get_resource(self, res_fn):
 118         offset = self._resources[res_fn]
 119         data = self._get_resource_data(offset)
 120         return data
 121
 122     def find_resource_names(self, res_fn_re):
 123         for key in self._resources.keys():
 124             m = re.match(res_fn_re, key)
 125             if m is not None:
 126                 yield key
 127
 128 class res_writer(object):
 129     def __init__(self, res):
 130         self.res = res
 131
 132     def _write_file(self, fn, data, decoder=None, zero_pad_to=None):
 133         if decoder is not None:
 134             data = decoder(data)
 135         if zero_pad_to is not None:
 136             if len(data) > zero_pad_to:
 137                 raise Exception("can not zero_pad_to %d bytes -- data is already %d bytes" % (zero_pad_to, len(data)))
 138             data += b"\0" * (zero_pad_to - len(data))
 139         with open(fn, "wb") as fp:
 140             fp.write(data)
 141         data_crc32 = zlib.crc32(data) & 0xffffffff
 142         print("saved %d bytes to %s (crc32=%08x)" % (len(data), fn, data_crc32))
 143
 144     def extract(self, res_fn, out_fn, decoder=None, zero_pad_to=None):
 145         self._write_file(out_fn, self.res.get_resource(res_fn), decoder=decoder, zero_pad_to=zero_pad_to)
 146
 147     def extract_re(self, res_fn_re, out_fn, decoder=None):
 148         for res_fn in res.find_resource_names(res_fn_re):
 149             fn = re.sub(res_fn_re, out_fn, res_fn).lower()
 150             self._write_file(fn, self.res.get_resource(res_fn), decoder=decoder)
 151
 152 def decode_intel_hex(hexdata):
 153     """ return list of (address, data)
 154     """
 155     datas = []
 156     # Assume LF-only or CR-LF style end-of-line.
 157     for line in hexdata.split(b"\n"):
 158         line = line.strip()
 159         if chr(line[0]) != ":": raise Exception("invalid line: %r" % line)
 160         offset = 1
 161         record = codecs.decode(line[offset:], "hex")
 162         byte_count, address, record_type = struct.unpack(">BHB", record[:1 + 2 + 1])
 163         offset = 1 + 2 + 1
 164         if byte_count > 0:
 165             data = record[offset:offset + byte_count]
 166             offset += byte_count
 167         checksum = record[offset]
 168         ex_checksum = (~sum(record[:offset]) + 1) & 0xff
 169         if ex_checksum != checksum: raise Exception("invalid checksum %#x in %r" % (checksum, line))
 170         if record_type == 0:
 171             datas.append((address, data))
 172         elif record_type == 1:
 173             break
 174     return datas
 175
 176 def intel_hex_as_blob(hexdata):
 177     """ return continuous bytes sequence including all data
 178     (loosing start address here)
 179     """
 180     data = decode_intel_hex(hexdata)
 181     data.sort()
 182     last = data[-1]
 183     length = last[0] + len(last[1])
 184     img = bytearray(length)
 185     for off, part in data:
 186         img[off:off + len(part)] = part
 187     return img
 188
 189 def maybe_intel_hex_as_blob(data):
 190     if data[0] == ord(":") and max(data) < 127:
 191           return intel_hex_as_blob(data)
 192     return data # Keep binary data.
 193
 194 if __name__ == "__main__":
 195     parser = argparse.ArgumentParser(description = "KingstVIS firmware extraction")
 196     parser.add_argument('executable', help = "KingstVIS executable file")
 197     options = parser.parse_args()
 198     exe_fn = options.executable
 199
 200     res = qt_resources(exe_fn)
 201
 202     writer = res_writer(res)
 203
 204     # extract all firmware and fpga bitstreams
 205     # writer.extract_re(r"fwfpga/(.*)", r"kingst-\1-fpga.bitstream")
 206     # writer.extract_re(r"fwusb/fw(.*)", r"kingst-la-\1.fw", decoder=maybe_intel_hex_as_blob)
 207
 208     # extract fx2 mcu firmware for both the la2016 and la1016
 209     # note that 0x01a2 is the usb pid for both of these devices
 210     writer.extract_re("fwusb/fw01A2", "kingst-la-01a2.fw", decoder=maybe_intel_hex_as_blob)
 211
 212     # extract fpga bitstreams for la2016
 213     # there are two bitstreams, newer hardware uses the 'a1' bitstream
 214     writer.extract_re("fwfpga/LA2016(.*)", r"kingst-la2016\1-fpga.bitstream")
 215
 216     # extract fpga bitstreams for la1016
 217     # there are two bitstreams, newer hardware uses the 'a1' bitstream
 218     writer.extract_re("fwfpga/LA1016(.*)", r"kingst-la1016\1-fpga.bitstream")