#!/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 lzo

import zlib

import os

import errno

import datetime

import sys

import shutil

import subprocess

import stat

import tempfile

from collections import defaultdict

try:

import zstandard as zstd

except ImportError:

zstd = None

if sys.version_info[0] == 2:

stdin = sys.stdin

stdout = sys.stdout

else:

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

# 3x

self.erasecount, # Q

self.vid_hdr_ofs, # L

self.data_ofs, # L

self.image_seq, # L

# 32x

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

# 4x

self.data_size, # L

self.used_ebs, # L

self.data_pad, # L

self.data_crc, # L

# 4x

self.sqnum, # Q

# 12x

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

# 23x

crc, # L

) = struct.unpack(">3LBBH128sB23xL", data)

if crc != crc32(data[:-4]):

raise Exception("crc mismatch")

self.name = self.name[:name_len]

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

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:

pass

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):

a = 0

for b in name:

if type(b)==str: b = ord(b)

a += b<<4

a += b>>4

a &= 0xFFFFFFFF

a *= 11

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

ip = 4

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

ip += 1

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:

break

if litlen:

if ip + 3 > len(data):

raise Exception("truncated LZ4K literals")

bits = (bits | (struct.unpack_from('<L', data, ip)[0] << remaining)) & 0xffffffff

ip += 3

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)

else:

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)

else:

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

# 4x

self.xattr_names, # L

self.compr_type # H

# 26x

) = 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

# 2x

)= 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

# x

self.type, # B

nlen, # H

# 4x

) = 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

# 12x

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):

(

# 2x

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

# 2x

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):