ubi.py

`ofrak.core.ubi`

`Ubi (GenericBinary)` `dataclass`

UBI is a volume management layer for raw / unmanaged flash devices. It can be thought of LVM (Logical Volume Manager) with some additional features necessary for reliably using raw flash memory such as wear leveling and error correction. Each volume can contain any arbitrary data, though UBIFS is specially made to be used within an UBI volume.

UBI parameters and volumes required by ubinize for repacking are defined here. Also see: http://www.linux-mtd.infradead.org/doc/ubi.html and https://github.com/vamanea/mtd-utils/blob/master/ubi-utils/ubinize.c#L288`

Attributes:

Name	Type	Description
`min_io_size`	`int`	Minimum number of bytes per I/O transaction (see http://www.linux-mtd.infradead.org/doc/ubi.html#L_min_io_unit)
`leb_size`	`int`	Size of Logical Erase Blocks
`peb_size`	`int`	Size of Physical Erase Blocks
`total_peb_count`	`int`	The total number of PEBs, which includes hidden layout blocks in addition to data blocks allocated per volume
`image_seq`	`int`	image sequence number recorded on EC headers (typically random)
`volumes`	`List[ofrak.core.ubi.UbiVolume]`	List of volumes associated with the UBI image

`UbiAnalyzer (Analyzer)`

Extract UBI parameters required for packing a resource.

`analyze(self, resource, config=None)` `async`

Analyze a resource for to extract specific ResourceAttributes.

Users should not call this method directly; rather, they should run Resource.run or Resource.analyze.

Parameters:

Name	Type	Description	Default
`resource`	`Resource`	The resource that is being analyzed	required
`config`		Optional config for analyzing. If an implementation provides a default, this default will always be used when config would otherwise be None. Note that a copy of the default config will be passed, so the default config values cannot be modified persistently by a component run.	`None`

Returns:

Type	Description
`Ubi`	The analysis results

Source code in ofrak/core/ubi.py

async def analyze(self, resource: Resource, config=None) -> Ubi:
    # Flush to disk
    with tempfile.NamedTemporaryFile() as temp_file:
        resource_data = await resource.get_data()
        temp_file.write(resource_data)
        temp_file.flush()

        ubi_obj = ubireader_ubi(
            ubi_io.ubi_file(
                temp_file.name,
                block_size=guess_peb_size(temp_file.name),
                start_offset=0,
                end_offset=None,
            )
        )

    # Technically multiple images can be encountered in an UBI blob, but that should be handled by
    # OFRAK by treating them as separate UBI resources.
    if len(ubi_obj.images) > 1:
        raise Exception(
            "Multi-image UBI blobs are not directly supported. Carve each image into a separate "
            "resource and run UbiAnalyzer on each of them."
        )
    if len(ubi_obj.images) == 0:
        raise Exception("UBI resource does not have any images.")

    image = ubi_obj.images[0]
    ubi_image_vols: List[UbiVolume] = []
    for volume in image.volumes.values():
        ubi_image_vols.append(
            UbiVolume(
                volume.vol_rec.rec_index,
                volume.vol_rec.reserved_pebs,
                PRINT_VOL_TYPE_LIST[volume.vol_rec.vol_type],
                volume.vol_rec.name.decode("utf-8"),
                volume.vol_rec.flags,  # Autoresize flag for standard UBI
                volume.vol_rec.alignment,
            )
        )

    return Ubi(
        ubi_obj.min_io_size,
        ubi_obj.leb_size,
        ubi_obj.peb_size,
        ubi_obj.block_count,
        image.image_seq,
        ubi_image_vols,
    )

`UbiIdentifier (Identifier)`

Check the first four bytes of a resource and tag the resource as Ubi if it matches the file magic.

`identify(self, resource, config=None)` `async`

Perform identification on the given resource.

Users should not call this method directly; rather, they should run Resource.identify.

Parameters:

Name	Type	Description	Default
`resource`	`Resource`		required
`config`		Optional config for identifying. If an implementation provides a default, this default will always be used when config would otherwise be None. Note that a copy of the default config will be passed, so the default config values cannot be modified persistently by a component run.	`None`

Source code in ofrak/core/ubi.py

async def identify(self, resource: Resource, config=None) -> None:
    datalength = await resource.get_data_length()
    if datalength >= 4:
        data = await resource.get_data(Range(0, 4))
        if data in [UBI_EC_HDR_MAGIC, UBI_VID_HDR_MAGIC]:
            resource.add_tag(Ubi)

`UbiPacker (Packer)`

Generate an UBI image from an Ubi resource view.

`pack(self, resource, config=None)` `async`

Pack the given resource.

Users should not call this method directly; rather, they should run Resource.run or Resource.pack.

Parameters:

Name	Type	Description	Default
`resource`	`Resource`		required
`config`		Optional config for packing. If an implementation provides a default, this default will always be used when config would otherwise be None. Note that a copy of the default config will be passed, so the default config values cannot be modified persistently by a component run.	`None`

Source code in ofrak/core/ubi.py

    async def pack(self, resource: Resource, config=None) -> None:
        ubi_view = await resource.view_as(Ubi)

        # with tempfile.NamedTemporaryFile(mode="rb") as temp:
        with tempfile.TemporaryDirectory() as temp_flush_dir:
            ubi_volumes = await resource.get_children()
            ubinize_ini_entries = []

            for volume in ubi_volumes:
                volume_view = await volume.view_as(UbiVolume)
                volume_size = await volume.get_data_length()

                # I think the `ubinize` rounds up the number of required PEBs based on the provided size.
                # Maybe this? allocated PEBs = -(volume_size // -peb_size) + 1
                # For empty volumes I reverse this operation
                if volume_size != 0:
                    volume_path = (
                        f"{temp_flush_dir}/input-{ubi_view.image_seq}_vol-{volume_view.name}.ubivol"
                    )
                    await volume.flush_data_to_disk(volume_path)
                else:
                    volume_path = None
                    volume_size = (volume_view.peb_count - 1) * ubi_view.peb_size

                # Generate a volume entry for `ubinize`'s config.ini
                ubinize_ini_entry = f"""\
[{volume_view.name}-volume]
mode=ubi
{(f"image={volume_path}" if volume_path else "")}
vol_id={volume_view.id}
vol_size={volume_size}
vol_type={volume_view.type}
vol_name={volume_view.name}
{(f"vol_flags=autoresize" if volume_view.flag_autoresize else "" )}
"""
                ubinize_ini_entries.append(ubinize_ini_entry)

            with open(f"{temp_flush_dir}/config.ini", "w") as config_ini_file:
                config_ini_file.write("\n".join(ubinize_ini_entries))

            cmd = [
                "ubinize",
                "-p",
                str(ubi_view.peb_size),
                "-m",
                str(ubi_view.min_io_size),
                "-o",
                f"{temp_flush_dir}/output.ubi",
                f"{temp_flush_dir}/config.ini",
            ]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
            )
            returncode = await proc.wait()
            if proc.returncode:
                raise CalledProcessError(returncode=returncode, cmd=cmd)

            with open(f"{temp_flush_dir}/output.ubi", "rb") as output_f:
                packed_blob_data = output_f.read()

            resource.queue_patch(Range(0, await resource.get_data_length()), packed_blob_data)

`UbiUnpacker (Unpacker)`

Extract the UBI image

`unpack(self, resource, config=None)` `async`

Unpack the given resource.

Users should not call this method directly; rather, they should run Resource.run or Resource.unpack.

Parameters:

Name	Type	Description	Default
`resource`	`Resource`	The resource that is being unpacked	required
`config`		Optional config for unpacking. If an implementation provides a default, this default will always be used when config would otherwise be None. Note that a copy of the default config will be passed, so the default config values cannot be modified persistently by a component run.	`None`

Source code in ofrak/core/ubi.py

async def unpack(self, resource: Resource, config=None):
    with tempfile.TemporaryDirectory() as temp_flush_dir:
        # flush to disk
        with open(f"{temp_flush_dir}/input.img", "wb") as temp_file:
            resource_data = await resource.get_data()
            temp_file.write(resource_data)
            temp_file.flush()

        # extract temp_file to temp_flush_dir
        cmd = [
            "ubireader_extract_images",
            "-o",
            f"{temp_flush_dir}/output",
            temp_file.name,
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
        )
        returncode = await proc.wait()
        if proc.returncode:
            raise CalledProcessError(returncode=returncode, cmd=cmd)

        ubi_view = await resource.view_as(Ubi)

        # Each file extracted by `ubireader_extract_images` is populated as an UbiVolume
        # `ubireader_extract_images` incorrectly appends a `ubifs` suffix despite unpacking ubi images / volumes
        for vol in ubi_view.volumes:
            f_path = (
                f"{temp_flush_dir}/output/{os.path.basename(temp_file.name)}"
                f"/img-{ubi_view.image_seq}_vol-{vol.name}.ubifs"
            )
            with open(f_path, "rb") as f:
                data = f.read()
                if len(data) > 0:
                    other_tags: Tuple[ResourceTag, ...] = (GenericBinary,)
                else:
                    other_tags = ()
                await resource.create_child_from_view(
                    vol, data=data, additional_tags=other_tags
                )

`UbiVolume (ResourceView)` `dataclass`

An UbiVolume is a volume entry in UBI. It contains an image of arbitrary data, typically a filesystem or a log. Empty UbiVolumes can still reserve physical erase blocks, in case they are expected to grow.

Volume information reflected in the 'config.ini' UBI volume entries expected by ubinize are stored here. Also see: http://www.linux-mtd.infradead.org/faq/ubi.html#L_ubi_mkimg

Attributes:

Name	Type	Description
`id`	`int`	The assigned volume ID within the UBI image
`peb_count`	`int`	Number of PEBs allocated for the volume
`type`	`str`	UBI volume type, either 'static' or 'dynamic'; see http://www.linux-mtd.infradead.org/doc/ubi.html#L_overview
`name`	`str`	Label assigned to the volume
`flag_autoresize`	`bool`	Tells UBI to resize this volume to occupy unused space in the UBI image once; see http://www.linux-mtd.infradead.org/doc/ubi.html#L_autoresize
`alignment`	`int`	LEB size of this volume has to be aligned on, such that Ubi.leb_size % UbiVolume.alignment == 0; see https://elixir.bootlin.com/linux/v6.1.7/source/include/uapi/mtd/ubi-user.h#L328

`_PyLzoTool (ComponentExternalTool)` `private`

`init(self)` `special`

Initialize self. See help(type(self)) for accurate signature.

Source code in ofrak/core/ubi.py

def __init__(self):
    super().__init__(
        "python-lzo",
        "https://github.com/jd-boyd/python-lzo",
        install_check_arg="",
    )

`is_tool_installed(self)` `async`

Check if a tool is installed by running it with the install_check_arg. This method runs <tool> <install_check_arg>.

Returns:

Type	Description
`bool`	True if the `tool` command returned zero, False if `tool` could not be found or returned non-zero exit code.

Source code in ofrak/core/ubi.py

async def is_tool_installed(self) -> bool:
    try:
        import lzo  # type: ignore

        return True
    except ModuleNotFoundError:
        return False

ubi.py

ofrak.core.ubi

Ubi (GenericBinary) dataclass

UbiAnalyzer (Analyzer)

analyze(self, resource, config=None) async

UbiIdentifier (Identifier)

identify(self, resource, config=None) async

UbiPacker (Packer)

pack(self, resource, config=None) async

UbiUnpacker (Unpacker)

unpack(self, resource, config=None) async

UbiVolume (ResourceView) dataclass

_PyLzoTool (ComponentExternalTool) private

__init__(self) special

is_tool_installed(self) async

`ofrak.core.ubi`

`Ubi (GenericBinary)` `dataclass`

`UbiAnalyzer (Analyzer)`

`analyze(self, resource, config=None)` `async`

`UbiIdentifier (Identifier)`

`identify(self, resource, config=None)` `async`

`UbiPacker (Packer)`

`pack(self, resource, config=None)` `async`

`UbiUnpacker (Unpacker)`

`unpack(self, resource, config=None)` `async`

`UbiVolume (ResourceView)` `dataclass`

`_PyLzoTool (ComponentExternalTool)` `private`

`init(self)` `special`

`is_tool_installed(self)` `async`