edl/ubidump at master · bkerler/edl

Repository navigation

#!/usr/bin/env python3
Tool for listing and extracting data from an UBI (Unsorted Block Image) image.
(C) 2017 by Willem Hengeveld <itsme@xs4all.nl>
(c) 2026 B.Kerler, additional mtk lz4 fixes
from __future__ import division, print_function
import crcmod.predefined
import argparse
import base64
import struct
from binascii import b2a_hex
import zlib
import errno
import datetime
import shutil
import subprocess
import stat
import tempfile
from collections import defaultdict
    import zstandard as zstd
except ImportError:
    zstd = None
if sys.version_info[0] == 2:
    stdin = sys.stdin
    stdout = sys.stdout
    stdin = sys.stdin.buffer
    stdout = sys.stdout.buffer
if sys.version_info[0] == 2:
    reload(sys)
    sys.setdefaultencoding('utf-8')
# Note that ubidump depends on the 'header=False' argument to compress/decompress,
# introduced in python-lzo 1.09
if sys.version_info[0] == 3:
    def cmp(a,b):
        return (a>b) - (a<b)
# using crcmod to generate the correct crc function.
crc32 = crcmod.predefined.mkPredefinedCrcFun('CrcJamCrc')
class SeekableStdout:
    Wrapper for stdout, which allows forward seeking.
    def __init__(self):
        self.pos = 0
    def seek(self, newpos, whence=os.SEEK_SET):
        if whence==os.SEEK_SET:
            if newpos < self.pos:
                print("WARNING: can't seek stdout backwards")
                return -1
            if newpos > self.pos:
                self.seekforward(newpos - self.pos)
            self.pos = newpos
        elif whence==os.SEEK_CUR:
            if newpos < 0:
                print("WARNING: can't seek stdout backwards")
                return -1
            if newpos > 0:
                self.seekforward(newpos)
            self.pos += newpos
        else:
            print("WARNING: can't seek stdout from EOF")
            return -1
    def seekforward(self, size):
        """
        Seek forward by writing NUL bytes.
        """
        sys.stdout.flush()
        chunk = b"\x00" * 0x10000
        while size > 0:
            if len(chunk) > size:
                chunk = chunk[:size]
            stdout.write(chunk)
            size -= len(chunk)
    def write(self, data):
        sys.stdout.flush()
        stdout.write(data)
        self.pos += len(data)
    def truncate(self, size):
        """
        Ignore this.
        """
        pass
########### block level objects ############
class UbiEcHeader:
    The Erase count header
    hdrsize = 16*4
    def __init__(self):
        self.magic = b'UBI#'
    def parse(self, data):
        self.magic,       # 4s
        self.version,     # B
        self.erasecount,  # Q
        self.vid_hdr_ofs, # L
        self.data_ofs,    # L
        self.image_seq,   # L
        hdr_crc,          # L
        ) = struct.unpack(">4sB3xQLLL32xL", data)
        if self.magic != b'UBI#':
            raise Exception("UBI# ec hdr magic num mismatch")
        if hdr_crc != crc32(data[:-4]):
            raise Exception("crc mismatch")
    def encode(self):
        data = struct.pack(">4sB3xQLLL32x", self.magic, self.version, self.erasecount, self.vid_hdr_ofs, self.data_ofs, \
                self.image_seq)
        return data + struct.pack(">L", crc32(data))
    def __repr__(self):
        return "EC: magic=%s, v%d, ec=%d, vidhdr=%x, data=%x, imgseq=%x" % ( 
                self.magic, self.version, self.erasecount, self.vid_hdr_ofs,
                self.data_ofs, self.image_seq)
VTBL_VOLID=0x7fffefff
class UbiVidHead:
    The volume id header
    hdrsize = 16*4
    def __init__(self):
        self.vol_id = VTBL_VOLID
        self.magic = b'UBI!'
    def parse(self, data):
        self.magic,        # 4s
        self.version,      # B
        self.vol_type,     # B
        self.copy_flag,    # B
        self.compat,       # B
        self.vol_id,       # L
        self.lnum,         # L
        self.data_size,    # L
        self.used_ebs,     # L
        self.data_pad,     # L
        self.data_crc,     # L
        self.sqnum,        # Q
        hdr_crc,           # L
        )= struct.unpack(">4s4BLL4x4L4xQ12xL", data)
        if self.magic != b'UBI!':
            raise Exception("UBI! volid magic num mismatch")
        if hdr_crc != crc32(data[:-4]):
            raise Exception("crc mismatch")
    def encode(self):
        data = struct.pack(">4s4BLL4x4L4xQ12x", self.magic, self.version, self.vol_type, self.copy_flag, self.compat, self.vol_id, \
                self.lnum, self.data_size, self.used_ebs, self.data_pad, self.data_crc, \
                self.sqnum)
        return data + struct.pack(">L", crc32(data))
    def __repr__(self):
        if hasattr(self, 'magic'):
            return "VID: magic=%s, v%d, vt=%d, cp=%d, compat=%d, volid=%x, lnum=[%d], " \
                    "dsize=%d, usedebs=%d, datapad=%d, datacrc=%x, sqnum=%d" % (
                            self.magic, self.version, self.vol_type, self.copy_flag, self.compat,
                            self.vol_id, self.lnum, self.data_size, self.used_ebs, self.data_pad,
                            self.data_crc, self.sqnum)
        else:
            return "VID"
class UbiVtblRecord:
    A volume table record.
    hdrsize = 4*4+128+24+4
    def __init__(self):
        self.reserved_pebs = 0
    def parse(self, data):
        self.reserved_pebs, # L
        self.alignment,     # L
        self.data_pad,      # L
        self.vol_type,      # B
        self.upd_marker,    # B
        name_len,           # H
        self.name,          # 128s
        self.flags,         # B
        ) = struct.unpack(">3LBBH128sB23xL", data)
        if crc != crc32(data[:-4]):
            raise Exception("crc mismatch")
        self.name = self.name[:name_len]<
BD0
/div>
    def encode(self):
        data = struct.pack(">3LBBH128sB23x", self.reserved_pebs, self.alignment, self.data_pad, self.vol_type, self.upd_marker, \
                name_len, self.name, self.flags)
        return data + struct.pack(">L", crc32(data))
    def empty(self):
        if hasattr(self, 'name'):
            return self.reserved_pebs==0 and self.alignment==0 and self.data_pad==0 \
                    and self.vol_type==0 and self.upd_marker==0 and self.name==b'' and self.flags==0
        else:
            return True

<
10000
/div>
    def __repr__(self):
        return "VREC: rsvpebs=%d, align=%d, datapad=%d, voltype=%d, updmark=%d, flags=%x, name=%s" % (
                self.reserved_pebs, self.alignment, self.data_pad, self.vol_type, 
                        self.upd_marker, self.flags, self.name)
class UbiVolume:
    provides read access to a specific volume in an UBI image.
    def __init__(self, blks, volid, dataofs):
        """
        takes an UbiBlocks object, a volumeid, and a baseoffset.
        """
        self.blks = blks
        self.volid = volid
        self.dataofs = dataofs
    def read(self, lnum, offs, size):
        return self.blks.readvolume(self.volid, lnum, self.dataofs+offs, size)
    def write(self, lnum, offs, data):
        return self.blks.writevolume(self.volid, lnum, self.dataofs+offs, data)
    def hexdump(self, lnum, offs, size):
        print("[%03d:0x%05x] %s" % (lnum, offs, b2a_hex(self.read(lnum, offs, size))))
    def saveraw(self, filename):
        with open(filename, "wb") as fh:
            for lnum in range(self.blks.maxlebs):
                data = self.read(lnum, 0, self.blks.leb_size-self.dataofs)
                fh.write(data)
class RawVolume:
    provides read access to a raw data volume
    def __init__(self, fh):
        self.fh = fh
        self.leb_size = self.find_block_size()
    def read(self, lnum, offs, size):
        self.fh.seek(lnum*self.leb_size+offs)
        return self.fh.read(size)
    def write(self, lnum, offs, data):
        self.fh.seek(lnum*self.leb_size+offs)
        return self.fh.write(data)
    def find_block_size(self):
        self.fh.seek(0)
        data = self.fh.read(0x200)
        values = struct.unpack("<12L", data[:4*12])
        if values[0] == 0x06101831 and values[5] == 6:
            # node magic, and nodetype == 6:superblock
            return values[9]  # sb.leb_size
    def hexdump(self, lnum, offs, size):
        print("R:[%03d:0x%05x] %s" % (lnum, offs, b2a_hex(self.read(lnum, offs, size))))
    def saveraw(self, filename):
        print("TODO")
class UbiBlocks:
    Block level access to an UBI image.
    def __init__(self, fh):
        self.fh = fh
        self.leb_size = self.find_blocksize()
        fh.seek(0, os.SEEK_END)
        self.filesize = fh.tell()
        self.maxlebs = self.filesize // self.leb_size
        self.scanblocks()
        if not VTBL_VOLID in self.vmap:
            print("no volume directory, %d physical volumes" % len(self.vmap))
            return
        self.scanvtbls(self.vmap[VTBL_VOLID][0])
        print("%d named volumes found, %d physical volumes, blocksize=0x%x" % (self.nr_named, len(self.vmap), self.leb_size))
    def find_blocksize(self):
        self.fh.seek(0)
        magic = self.fh.read(4)
        if magic != b'UBI#':
            raise Exception("not an UBI image")
        for log_blocksize in range(10,20):
            self.fh.seek(1<<log_blocksize)
            magic = self.fh.read(4)
            if magic == b'UBI#':
                return 1<<log_blocksize
        raise Exception("Could not determine UBI image blocksize")
    def scanblocks(self):
        """
        creates map of volid + lnum => physical lnum
        """
        self.vmap = defaultdict(lambda : defaultdict(int))
        for lnum in range(self.maxlebs):
            try:
                ec = UbiEcHeader()
                hdr = self.readblock(lnum, 0, ec.hdrsize)
                ec.parse(hdr)
                vid = UbiVidHead()
                viddata = self.readblock(lnum, ec.vid_hdr_ofs, vid.hdrsize)
                vid.parse(viddata)
                self.vmap[vid.vol_id][vid.lnum] = lnum
            except:
    def readblock(self, lnum, offs, size):
        self.fh.seek(lnum * self.leb_size + offs)
        return self.fh.read(size)
    def writeblock(self, lnum, offs, data):
        self.fh.seek(lnum * self.leb_size + offs)
        return self.fh.write(data)
    def hexdump(self, lnum, offs, size):
        print("[%03d:0x%05x] %s" % (lnum, offs, b2a_hex(self.readblock(lnum, offs, size))))
    def scanvtbls(self, lnum):
        """
        reads the volume table
        """
        ec = UbiEcHeader()
        hdr = self.readblock(lnum, 0, ec.hdrsize)
        ec.parse(hdr)
        self.ec = ec
        try:
            vid = UbiVidHead()
            viddata = self.readblock(lnum, ec.vid_hdr_ofs, vid.hdrsize)
            vid.parse(viddata)
            self.vid = vid
            self.vtbl = []
            self.nr_named = 0
            if vid.vol_id == VTBL_VOLID:
                for i in range(128):
                    vrec = UbiVtblRecord()
                    vrecdata = self.readblock(lnum, self.ec.data_ofs + i * vrec.hdrsize, vrec.hdrsize)
                    vrec.parse(vrecdata)
                    self.vtbl.append(vrec)
                    if not vrec.empty():
                        self.nr_named += 1
        except:
            print(ec)
            print("viddata:%s" % b2a_hex(viddata))
            import traceback
            traceback.print_exc()
            self.vid = UbiVidHead()
            self.vtbl = [ UbiVtblRecord() ]
    def dumpvtbl(self):
        print("%s  %s" % (self.ec, self.vid))
        for v in self.vtbl:
            if not v.empty():
                print("  %s" % v)
        for volid, lmap in self.vmap.items():
            print("volume %x : %d lebs" % (volid, len(lmap)))
    def nr_named(self):
        return self.nr_named
    def getvrec(self, volid):
        return self.vtbl[volid]
    def getvolume(self, volid):
        return UbiVolume(self, volid, self.ec.data_ofs)
    def readvolume(self, volid, lnum, offs, size):
        physlnum = self.vmap[volid].get(lnum, None)
        if physlnum is None:
            raise Exception("volume does not contain lnum")
        return self.readblock(physlnum, offs, size)
    def writevolume(self, volid, lnum, offs, data):
        physlnum = self.vmap[volid].get(lnum, None)
        if physlnum is None:
            raise Exception("volume does not contain lnum")
        return self.writeblock(physlnum, offs, data)
################ filesytem level objects ##################
UBIFS_INO_KEY = 0
UBIFS_DATA_KEY = 1
UBIFS_DENT_KEY = 2
UBIFS_XENT_KEY = 3
key format:  (inum, (type<<29) | value)
key types: UBIFS_*_KEY: INO, DATA, DENT, XENT
inode:  <inum>  + 0
dirent:  <inum>  + hash
xent:  <inum>  + hash
data:  <inum + blocknum
trunc:  <inum>  + 0
def unpackkey(key):
    if len(key)==16 and key[8:]!=b'\x00'*8:
        print("key has more than 8 bytes: %s" % b2a_hex(key))
    inum, value = struct.unpack("<LL", key[:8])
    return (inum, value>>29, value&0x1FFFFFFF)
def packkey(key):
    inum, ityp, value = key
    return struct.pack("<LL", inum, (ityp<<29) | value)
def formatkey(key):
    if key is None:
        return "None"
    if type(key) != tuple:
        key = unpackkey(key)
    return "%05d:%d:%08x" % key
def comparekeys(lhs, rhs):
    return cmp(unpackkey(lhs), unpackkey(rhs))
def namehash(name):
    for b in name:
        if type(b)==str: b = ord(b)
        a += b<<4
        a += b>>4
        a &= 0xFFFFFFFF
        a &= 0xFFFFFFFF
    a &= 0x1FFFFFFF
    if a <= 2: a += 3
    return a
COMPR_NONE = 0
COMPR_LZO = 1
COMPR_ZLIB = 2
COMPR_ZSTD = 3
# MediaTek's UBIFS fork uses compression type 3 for its LZ4K compressor.
# LZ4K is a bit-packed LZ decoder and its payloads end in the ASCII tag LZ4K;
# it is not compatible with the normal LZ4 or Zstandard formats.
LZ4K_TAG = b'LZ4K'
def _lz4k_table(encoded, count):
    return struct.unpack("<%dH" % count, base64.b64decode(encoded))
_lz4k_matchlen_decode = _lz4k_table(
    "DAMSBQsEAAAJBBEGEAQAAAwDGAUPBAAACgQfBg0FAAAMAxQFCwQAAAkEFgYQBAAADAMcBQ8EAAAKBBoHDgUAAAwDEgULBAAACQQTBhAEAAAMAxgFDwQAAAoEFQcNBQAADAMUBQsEAAAJBBcGEAQAAAwDHAUPBAAACgQdBw4FAAAMAxIFCwQAAAkEEQYQBAAADAMYBQ8EAAAKBB8GDQUAAAwDFAULBAAACQQWBhAEAAAMAxwFDwQAAAoEGwcOBQAADAMSBQsEAAAJBBMGEAQAAAwDGAUPBAAACgQZBw0FAAAMAxQFCwQAAAkEFwYQBAAADAMcBQ8EAAAKBB4HDgUAAA==", 128)
_lz4k_matchoff_decode = _lz4k_table(
    "BAQ4BRAEQAYIBCAGDAVgBwQEFAYQBDwHCAQoBhgFVAgEBDgFEAQsBwgEJAYMBXgHBAQcBhAEUAcIBDAGGAVsCAQEOAUQBEAGCAQgBgwFcAcEBBQGEARIBwgEKAYYBWQIBAQ4BRAENAcIBCQGDAVECAQEHAYQBFgHCAQwBhgFfAgEBDgFEARABggEIAYMBWAHBAQUBhAEPAcIBCgGGAVcCAQEOAUQBCwHCAQkBgwFeAcEBBwGEARQBwgEMAYYBXQIBAQ4BRAEQAYIBCAGDAVwBwQEFAYQBEgHCAQoBhgFaAgEBDgFEAQ0BwgEJAYMBUwIBAQcBhAEWAcIBDAGGAWACA==", 128)
_lz4k_literallen_decode = _lz4k_table(
    "AQECAwEBBQQBAQMEAQEHBgEBAgMBAQYFAQEEBAEBDgcBAQIDAQEFBAEBAwQBAQoHAQECAwEBCQUBAQQEAQEAAAEBAgMBAQUEAQEDBAEBCAYBAQIDAQEGBQEBBAQBARAHAQECAwEBBQQBAQMEAQEMBwEBAgMBAQkFAQEEBAEBAAABAQIDAQEFBAEBAwQBAQcGAQECAwEBBgUBAQQEAQEPBwEBAgMBAQUEAQEDBAEBCwcBAQIDAQEJBQEBBAQBAQAAAQECAwEBBQQBAQMEAQEIBgEBAgMBAQYFAQEEBAEBEQcBAQIDAQEFBAEBAwQBAQ0HAQECAwEBCQUBAQQEAQEAAA==", 128)
_lz4k_literalch_decode = _lz4k_table(
    "AARICDAHFwlABbAIDAidCf8EZwhvB2sJAgfoCCIIxAkABFQIYQc/CQEG0ggYCK8J/wR4CIAHiwkJB/QIMQjcCQAETghQBzYJQAXACBEIpgn/BHEIcgeBCQQH8AgpCM0JAARiCGkHVwkgBtgIHQi6Cf8EkAgFCJQJEAf5CDgI6QkABEwIQQcqCUAFuAgOCKIJ/wRsCHAHfAkDB+0IJAjJCQAEXwhlB0kJAQbUCBsItQn/BIQI4AePCQoH9wgzCOIJAARSCFEHOwlABdAIEwirCf8EdQh0B4YJCAfyCC4I1wkABGQIbgdcCSAG4wgfCL8J/wSgCAcImQkoB/wIQwj2CQAESwgwByYJQAW0CA0Inwn/BGgIbwd6CQIH7AgjCMYJAARYCGEHRgkBBtMIGgiyCf8EeQiAB40JCQf1CDII3gkABE8IUAc5CUAFyAgSCKkJ/wRzCHIHgwkEB/EILAjPCQAEYwhpB1oJIAbhCB4IvQn/BJgIBgiWCRAH+ghCCOsJAARNCEEHLQlABbwIDwikCf8EbQhwB34JAwfuCCUIywkABGAIZQdVCQEG1QgcCLcJ/wSICOAHkgkKB/gINAjmCQAEUwhRBz0JQAXRCBQIrQn/BHYIdAeJCQgH8wgvCNoJAARmCG4HXgkgBuUIIQjCCf8EqAgLCJsJKAcVCUQI/QkABEgIMAcZCUAFsAgMCJ4J/wRnCG8HdwkCB+gIIgjFCQAEVAhhB0UJAQbSCBgIsQn/BHgIgAeMCQkH9AgxCN0JAAROCFAHNwlABcAIEQinCf8EcQhyB4IJBAfwCCkIzgkABGIIaQdZCSAG2AgdCLsJ/wSQCAUIlQkQB/kIOAjqCQAETAhBBysJQAW4CA4Iown/BGwIcAd9CQMH7QgkCMoJAARfCGUHSgkBBtQIGwi2Cf8EhAjgB5EJCgf3CDMI5AkABFIIUQc8CUAF0AgTCKwJ/wR1CHQHhwkIB/IILgjZCQAEZAhuB10JIAbjCB8IwQn/BKAIBwiaCSgH/AhDCPsJAARLCDAHJwlABbQIDQihCf8EaAhvB3sJAgfsCCMIxwkABFgIYQdHCQEG0wgaCLMJ/wR5CIAHjgkJB/UIMgjfCQAETwhQBzoJQAXICBIIqgn/BHMIcgeFCQQH8QgsCNYJAARjCGkHWwkgBuEIHgi+Cf8EmAgGCJcJEAf6CEII7wkABE0IQQc1CUAFvAgPCKUJ/wRtCHAHfwkDB+4IJQjMCQAEYAhlB1YJAQbVCBwIuQn/BIgI4AeTCQoH+Ag0COcJAARTCFEHPglABdEIFAiuCf8Edgh0B4oJCAfzCC8I2wkABGYIbgdqCSAG5QghCMMJ/wSoCAsInAkoBxYJRAj+CQ==", 512)
def lz4k_decompress(data, buflen):
    """Decode a MediaTek LZ4K UBIFS data block."""
    if buflen > 4096 or len(data) < 8 or data[-4:] != LZ4K_TAG:
        raise Exception("invalid LZ4K data")
    bits = struct.unpack_from('<L', data, 0)[0] >> 1
    remaining = 31
    output = bytearray()
    def reserve(bits, remaining, ip):
        if remaining <= 16:
            if ip + 2 > len(data):
                raise Exception("truncated LZ4K bitstream")
            bits = (bits | (struct.unpack_from('<H', data, ip)[0] << remaining)) & 0xffffffff
            return bits, remaining + 16, ip + 2
        return bits, remaining, ip
    def emit_match(offset, length):
        if offset <= 0 or offset > len(output) or len(output) + length > buflen:
            raise Exception("invalid LZ4K match")
        for _ in range(length):
            output.append(output[-offset])
    while True:
        bits, remaining, ip = reserve(bits, remaining, ip)
        if (bits & 1) == 0:
            if (bits & 2) == 0:
                value = _lz4k_literalch_decode[(bits >> 2) & 0x1ff]
                consumed = value >> 8
                output.append(value & 0xff)
                bits >>= consumed + 2
                remaining -= consumed + 2
            else:
                value = _lz4k_literallen_decode[(bits >> 1) & 0x7f]
                if value:
                    consumed = value >> 8
                    litlen = value & 0xff
                    bits >>= consumed + 1
                    remaining -= consumed + 1
                else:
                    if remaining < 18:
                        if ip >= len(data):
                            raise Exception("truncated LZ4K literal length")
                        bits |= data[ip] << remaining
                        remaining += 8
                    litlen = ((bits >> 6) & 0xfff) + 1
                    bits >>= 18
                    remaining -= 18
                while litlen:
                    while remaining > 8:
                        value = _lz4k_literalch_decode[bits & 0x1ff]
                        consumed = value >> 8
                        output.append(value & 0xff)
                        bits >>= consumed
                        remaining -= consumed
                        litlen -= 1
                        if not litlen:
                    if litlen:
                        if ip + 3 > len(data):
                            raise Exception("truncated LZ4K literals")
                        bits = (bits | (struct.unpack_from('<L', data, ip)[0] << remaining)) & 0xffffffff
                        remaining += 24
            if len(output) == buflen:
                break
        else:
            bits >>= 1
            remaining -= 1
        bits, remaining, ip = reserve(bits, remaining, ip)
        if (bits & 1) == 0:
            value = _lz4k_matchoff_decode[(bits & 0xff) >> 1]
            consumed = value >> 8
            offset = value & 0xff
            bits >>= consumed
            remaining -= consumed
        else:
            offset = (bits >> 1) & 0xfff
            bits >>= 13
            remaining -= 13
        bits, remaining, ip = reserve(bits, remaining, ip)
        if (bits & 1) == 0:
            length = 3 + ((bits >> 1) & 1)
            bits >>= 2
            remaining -= 2
        elif (bits & 2) == 0:
            length = 5 + ((bits >> 2) & 3)
            bits >>= 4
            remaining -= 4
        else:
            value = _lz4k_matchlen_decode[(bits >> 2) & 0x7f]
            if value:
                consumed = value >> 8
                length = value & 0xff
                bits >>= consumed + 2
                remaining -= consumed + 2
            else:
                length = (bits >> 4) & 0xfff
                bits >>= 16
                remaining -= 16
        emit_match(offset, length)
        if len(output) == buflen:
            break
    if len(output) != buflen:
        raise Exception("LZ4K data size mismatch")
    return bytes(output)
def decompress(data, buflen, compr_type):
    if compr_type==COMPR_NONE:
        return data
    elif compr_type==COMPR_LZO:
        return lzo.decompress(data, False, buflen)
    elif compr_type==COMPR_ZLIB:
        return zlib.decompress(data, -zlib.MAX_WBITS)
    elif compr_type==COMPR_ZSTD and data.endswith(LZ4K_TAG):
        return lz4k_decompress(data, buflen)
    elif compr_type==COMPR_ZSTD and zstd:
        # UBIFS stores the uncompressed size in the data-node header.  Do not
        # rely on a Zstandard content-size field: kernel-produced streaming
        # frames may omit it.
        dctx = zstd.ZstdDecompressor()
        try:
            return dctx.decompress(data, max_output_size=buflen)
        except zstd.ZstdError:
            return dctx.decompressobj().decompress(data)
        raise Exception("unknown compression type")
def compress(data, compr_type):
    if compr_type==COMPR_NONE:
        return data
    elif compr_type==COMPR_LZO:
        return lzo.compress(data, False)
    elif compr_type==COMPR_ZLIB:
        return zlib.compress(data, -zlib.MAX_WBITS)
    elif compr_type==COMPR_ZSTD and zstd:
        return zstd.compress(data)
        raise Exception("unknown compression type")
# the blocksize is a fixed value, independent of the underlying device.
UBIFS_BLOCKSIZE = 4096
########### objects for the various node types ########### 
class UbiFsInode:
    Leafnode in the B-tree, contains information for a specific file or directory.
    It's b-tree key is formatted like this:
       * 32 bit inode number
       * the 3 bit node type: 0 for inode
       * a 29 bit zero value.
    nodetype = 0
    hdrsize = 16 + 5*8 + 11*4 + 2*4 + 28
    # note: these values are like the posix stat values,
    # the UbiFsDirEntry uses a different set of values for the same types.
    ITYPE_FIFO      =  1  # S_IFIFO 
    ITYPE_CHARDEV   =  2  # S_IFCHR 
    ITYPE_DIRECTORY =  4  # S_IFDIR 
    ITYPE_BLOCKDEV  =  6  # S_IFBLK 
    ITYPE_REGULAR   =  8  # S_IFREG 
    ITYPE_SYMLINK   = 10  # S_IFLNK 
    ITYPE_SOCKET    = 12  # S_IFSOCK
    def __init__(self):
        pass
    def parse(self, data):
        self.key,          # 16s
        self.creat_sqnum,  # Q
        self.size,         # Q
        self.atime_sec,    # Q
        self.ctime_sec,    # Q
        self.mtime_sec,    # Q
        self.atime_nsec,   # L
        self.ctime_nsec,   # L
        self.mtime_nsec,   # L
        self.nlink,        # L
        self.uid,          # L
        self.gid,          # L
        self.mode,         # L
        self.flags,        # L
        self.data_len,     # L
        self.xattr_cnt,    # L
        self.xattr_size,   # L
        self.xattr_names,  # L
        self.compr_type    # H
        ) = struct.unpack("<16s5Q11L4xLH26x", data[:self.hdrsize])
        # data contains the symlink string for symbolic links
        self.data = data[self.hdrsize:]
        if len(self.data) != self.data_len:
            raise Exception("inode data size mismatch")
    def encode(self):
        return struct.pack("<16s5Q11L4xLH26x", \
                self.key, self.creat_sqnum, self.size, self.atime_sec, self.ctime_sec, self.mtime_sec, \
                self.atime_nsec, self.ctime_nsec, self.mtime_nsec, self.nlink, self.uid, self.gid, \
                self.mode, self.flags, self.data_len, self.xattr_cnt, self.xattr_size, \
                self.xattr_names, self.compr_type)
    def inodedata_repr(self):
        types = ["0", "FIFO", "CHAR", "3", "DIRENT", "5", "BLOCK", "7", "FILE", "9", "LINK", "11", "SOCK", "13", "14", "15"]
        typ = self.nodetype()
        if typ in (self.ITYPE_CHARDEV, self.ITYPE_BLOCKDEV):  # CHAR or BLOCK
            return types[typ] + ":" + b2a_hex(self.data).decode('ascii')
        return types[typ] + ":%s" % self.data
    def __repr__(self):
        return "INODE: key=%s, sq=%04x, size=%5d, n=%3d, uid:gid=%d:%d, mode=%06o, fl=%x, dl=%3d, " \
                "xattr=%d:%d, xanames=%d, comp=%d -- %s" % (formatkey(self.key), self.creat_sqnum, 
                        self.size, self.nlink, self.uid, self.gid, self.mode, self.flags, self.data_len, 
                        self.xattr_cnt, self.xattr_size, self.xattr_names, self.compr_type,  self.inodedata_repr())
        # todo: self.atime_sec, self.ctime_sec, self.mtime_sec, self.atime_nsec, self.ctime_nsec, self.mtime_nsec, 
    def atime(self):
        return self.atime_sec + self.atime_nsec / 1000000000.0
    def mtime(self):
        return self.mtime_sec + self.mtime_nsec / 1000000000.0
    def ctime(self):
        return self.ctime_sec + self.ctime_nsec / 1000000000.0
    def devnum(self):
        ma, mi = struct.unpack("BB", self.data[:2])
        return (ma, mi)
    def nodetype(self):
        return (self.mode >> 12) & 0xF
class UbiFsData:
    Leafnode in the B-tree, contains a datablock
    It's b-tree key is formatted like this:
       * 32 bit inode number
       * the 3 bit node type: 1 for data
       * a 29 bit file blocknumber
    nodetype = 1
    hdrsize = 16 + 4 + 4
    def __init__(self):
        pass
    def parse(self, data):
        self.key,        # 16s
        self.size,       # L
        self.compr_type, # H
        )= struct.unpack("<16sLH2x", data[:self.hdrsize])
        compressed_data = data[self.hdrsize:]
        self.undecoded = False
        try:
            self.data = decompress(compressed_data, self.size, self.compr_type)
        except Exception as exc:
            # Some vendor UBIFS images label a non-standard/encrypted payload
            # as ZSTD.  Keep the payload available so extraction can continue;
            # callers get the original bytes plus an explicit warning instead
            # of losing the remainder of the filesystem.
            if self.compr_type != COMPR_ZSTD:
                raise
            self.data = compressed_data
            self.undecoded = True
            self.compression_error = exc
        if not self.undecoded and len(self.data) != self.size:
            raise Exception("data size mismatch")
    def encode(self):
        return struct.pack("<16sLH2x", self.key, len(self.data), self.compr_type) + compress(self.data, self.compr_type)
    def __repr__(self):
        return "DATA: key=%s, size=%d, comp=%d" % (formatkey(self.key), self.size, self.compr_type)
class UbiFsDirEntry:
    Leafnode in the B-tree, contains a directory entry.
    Properties:
      * key
      * inum
      * type
      * name
    It's b-tree key is formatted like this:
       * 32 bit inode number ( of the directory containing this dirent )
       * the 3 bit node type: 2 for dirent
       * a 29 bit name hash
    TYPE_REGULAR = 0
    TYPE_DIRECTORY = 1
    TYPE_SYMLINK = 2
    TYPE_BLOCKDEV = 3
    TYPE_CHARDEV = 4
    TYPE_FIFO = 5
    TYPE_SOCKET = 6
    ALL_TYPES = 127
    nodetype = 2
    hdrsize = 16 + 8+4+4
    def __init__(self):
        pass
    def parse(self, data):
            self.key,  # 16s
            self.inum, # Q
            self.type, # B
            nlen,      # H
        ) = struct.unpack("<16sQxBH4x", data[:self.hdrsize])
        self.name = data[self.hdrsize:-1]
        if len(self.name) != nlen:
            raise Exception("name length mismatch")
    def encode(self):
        return struct.pack("<16sQxBH4x", self.key, self.inum, self.type, nlen)
    def __repr__(self):
        typenames = [ 'reg', 'dir', 'lnk', 'blk', 'chr', 'fifo', 'sock' ]
        # type: UBIFS_ITYPE_REG, UBIFS_ITYPE_DIR, etc
        return "DIRENT: key=%s, inum=%05d, type=%d:%s -- %s" % (formatkey(self.key), self.inum, self.type, typenames[self.type], self.name)
class UbiFsExtendedAttribute:
    Leafnode in the B-tree, contains extended attributes.
    It's b-tree key is formatted like this:
       * 32 bit inode number ( of the directory containing this dirent )
       * the 3 bit node type: 3 for xent
       * a 29 bit hash of the attribute name.
    nodetype = 3
    hdrsize = 0
    def __init__(self):
        pass
    def parse(self, data):
        # TODO
        pass
    def __repr__(self):
        return "EA"
class UbiFsTruncation:
    Used only in the journal
    nodetype = 4
    hdrsize = 4+12+2*8
    def __init__(self):
        pass
    def parse(self, data):
            self.inum,     # L
            self.old_size, # Q
            self.new_size, # Q
        ) = struct.unpack("<L12xQQ", data)
    def encode(self):
        return struct.pack("<L12xQQ", self.inum, self.old_size, self.new_size)
    def __repr__(self):
        return "TRUNC: inum:%05d, size:%d->%d" % (self.inum, self.old_size, self.new_size)
class UbiFsPadding:
    nodetype = 5
    hdrsize = 4
    def __init__(self):
        pass
    def parse(self, data):
        self.pad_len, = struct.unpack_from("<L", data, 0)
    def encode(self):
        return struct.pack("<L", self.pad_len)
    def __repr__(self):
        return "PAD: padlen=%d" % self.pad_len
class UbiFsSuperblock:
    This object can be referenced via UbiFs.sb
    nodetype = 6
    hdrsize = 6*4+8+7*4+3*4+8+4+16+4
    def __init__(self):
        pass
    def parse(self, data):
        self.key_hash,              # B
        self.key_fmt,               # B
        self.flags,                 # L
        self.min_io_size,           # L
        self.leb_size,              # L
        self.leb_cnt,               # L
        self.max_leb_cnt,           # L
        self.max_bud_bytes,         # Q
        self.log_lebs,              # L
        self.lpt_lebs,              # L
        self.orph_lebs,             # L
        self.jhead_cnt,             # L
        self.fanout,                # L
        self.lsave_cnt,             # L
        self.fmt_version,           # L
        self.default_compr,         # H
        self.rp_uid,                # L
        self.rp_gid,                # L
        self.rp_size,               # Q
        self.time_gran,             # L
        self.uuid,                  # 16s
        self.ro_compat_version,     # L
        ) = struct.unpack("<2xBB5LQ7LH2xLLQL16sL", data[:self.hdrsize])
        if len(data) != self.hdrsize + 3968:
            raise Exception("invalid superblock padding size")
    def encode(self):
        return struct.pack("<2xBB5LQ7LH2xLLQL16sL", 
                self.key_hash, self.key_fmt, self.flags, self.min_io_size, self.leb_size, self.leb_cnt, \
                self.max_leb_cnt, self.max_bud_bytes, self.log_lebs, self.lpt_lebs, self.orph_lebs, \
                self.jhead_cnt, self.fanout, self.lsave_cnt, self.fmt_version, self.default_compr, \
                self.rp_uid, self.rp_gid, self.rp_size, self.time_gran, self.uuid, self.ro_compat_version)
    def __repr__(self):
        return "SUPER: kh:%d, fmt:%d, flags=%x, minio=%d, lebsize=0x%x, lebcount=%d, maxleb=%d, " \
                "maxbud=%d, loglebs=%d, lptlebs=%d, orphlebs=%d, jheads=%d, fanout=%d, lsave=%d, " \
                "fmt=v%d, compr=%d, rp=%d:%d, rpsize=%d, timegran=%d, uuid=%s, rocompat=%d" % (
                        self.key_hash, self.key_fmt, self.flags, self.min_io_size, self.leb_size,
                        self.leb_cnt, self.max_leb_cnt, self.max_bud_bytes, self.log_lebs, self.lpt_lebs,
                        self.orph_lebs, self.jhead_cnt, self.fanout, self.lsave_cnt, self.fmt_version,
                        self.default_compr, self.rp_uid, self.rp_gid, self.rp_size, self.time_gran,
                        b2a_hex(self.uuid), self.ro_compat_version)
class UbiFsMaster:
    This object can be referenced via UbiFs.mst
    nodetype = 7
    hdrsize = 2*8+8*4+6*8+12*4
    def __init__(self):
        pass
    def parse(self, data):
        self.highest_inum, # Q
        self.cmt_no,       # Q
        self.flags,        # L
        self.log_lnum,     # L
        self.root_lnum,    # L
        self.root_offs,    # L
        self.root_len,     # L
        self.gc_lnum,      # L
        self.ihead_lnum,   # L
        self.ihead_offs,   # L
        self.index_size,   # Q
        self.total_free,   # Q
        self.total_dirty,  # Q
        self.total_used,   # Q
        self.total_dead,   # Q
        self.total_dark,   # Q
        self.lpt_lnum,     # L
        self.lpt_offs,     # L
        self.nhead_lnum,   # L
        self.nhead_offs,   # L
        self.ltab_lnum,    # L
        self.ltab_offs,    # L
        self.lsave_lnum,   # L
        self.lsave_offs,   # L
        self.lscan_lnum,   # L
        self.empty_lebs,   # L
        self.idx_lebs,     # L
        self.leb_cnt,      # L
        ) = struct.unpack("<QQ8L6Q12L", data[:self.hdrsize])
        if len(data) != self.hdrsize + 344:
            raise Exception("invalid master padding size")
    def encode(self):
        return struct.pack("<QQ8L6Q12L", self.highest_inum, self.cmt_no, self.flags, self.log_lnum, self.root_lnum, self.root_offs, \
                self.root_len, self.gc_lnum, self.ihead_lnum, self.ihead_offs, self.index_size, \
                self.total_free, self.total_dirty, self.total_used, self.total_dead, \
                self.total_dark, self.lpt_lnum, self.lpt_offs, self.nhead_lnum, self.nhead_offs, \
                self.ltab_lnum, self.ltab_offs, self.lsave_lnum, self.lsave_offs, self.lscan_lnum, \
                self.empty_lebs, self.idx_lebs, self.leb_cnt)
    def __repr__(self):
View remainder of file in raw view
FilesExpand file tree

ubidump

Latest commit

History

ubidump

File metadata and controls
