squashfs.py

ofrak.core.squashfs

SquashfsFilesystem (GenericBinary, FilesystemRoot) dataclass

Filesystem stored in a squashfs format.

SquashfsPacker (Packer)

Pack files into a compressed squashfs filesystem.

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/squashfs.py

async def pack(self, resource: Resource, config=None):
    squashfs_view: SquashfsFilesystem = await resource.view_as(SquashfsFilesystem)
    temp_flush_dir = await squashfs_view.flush_to_disk()
    with tempfile.NamedTemporaryFile(suffix=".sqsh", mode="rb") as temp:
        cmd = [
            "mksquashfs",
            temp_flush_dir,
            temp.name,
            "-noappend",
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
        )
        returncode = await proc.wait()
        if proc.returncode:
            raise CalledProcessError(returncode=returncode, cmd=cmd)
        new_data = temp.read()
        # Passing in the original range effectively replaces the original data with the new data
        resource.queue_patch(Range(0, await resource.get_data_length()), new_data)

SquashfsUnpacker (Unpacker)

Unpack a SquashFS filesystem.

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/squashfs.py

async def unpack(self, resource: Resource, config=None):
    with tempfile.NamedTemporaryFile() as temp_file:
        resource_data = await resource.get_data()
        temp_file.write(resource_data)
        temp_file.flush()

        with tempfile.TemporaryDirectory() as temp_flush_dir:
            cmd = [
                "unsquashfs",
                "-no-exit-code",
                "-force",
                "-dest",
                temp_flush_dir,
                temp_file.name,
            ]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
            )
            returncode = await proc.wait()
            if proc.returncode:
                raise CalledProcessError(returncode=returncode, cmd=cmd)

            squashfs_view = await resource.view_as(SquashfsFilesystem)
            await squashfs_view.initialize_from_disk(temp_flush_dir)

_UnsquashfsV45Tool (ComponentExternalTool) private

__init__(self) special

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

Source code in ofrak/core/squashfs.py

def __init__(self):
    super().__init__("unsquashfs", "https://github.com/plougher/squashfs-tools", "")

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/squashfs.py

async def is_tool_installed(self) -> bool:
    try:
        cmd = ["unsquashfs", "-help"]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        stdout, stderr = await proc.communicate()
    except FileNotFoundError:
        return False

    if 0 != proc.returncode:
        return False

    if b"-no-exit" not in stdout:
        # Version 4.5+ has the required -no-exit option
        return False

    return True