zip.py

ofrak.core.zip

ZipArchive (GenericBinary, FilesystemRoot) dataclass

Filesystem stored in a zip archive.

ZipPacker (Packer)

Compresses and packages files into a ZIP archive format with standard compression algorithms. Use after modifying extracted ZIP contents to recreate the archive for distribution or deployment. The packer preserves directory structure and file attributes during compression.

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

async def pack(self, resource: Resource, config=None):
    zip_view: ZipArchive = await resource.view_as(ZipArchive)
    flush_dir = await zip_view.flush_to_disk()
    temp_archive = f"{flush_dir}.zip"
    cwd = os.getcwd()
    os.chdir(flush_dir)
    cmd = [
        "zip",
        "-r",
        temp_archive,
        ".",
    ]
    proc = await asyncio.create_subprocess_exec(
        *cmd,
    )
    returncode = await proc.wait()
    if proc.returncode:
        raise CalledProcessError(returncode=returncode, cmd=cmd)
    os.chdir(cwd)
    with open(temp_archive, "rb") as fh:
        resource.queue_patch(Range(0, await zip_view.resource.get_data_length()), fh.read())

ZipUnpacker (Unpacker)

Extracts files and directories from ZIP compressed archives using standard ZIP decompression. Use for any ZIP-packaged software distributions, firmware bundles, or data archives. After extraction, individual files can be analyzed, modified, and later repacked with ZipPacker.

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

async def unpack(self, resource: Resource, config=None):
    zip_view = await resource.view_as(ZipArchive)
    async with resource.temp_to_disk(suffix=".zip") as temp_path:
        with tempfile.TemporaryDirectory() as temp_dir:
            cmd = [
                "unzip",
                temp_path,
                "-d",
                temp_dir,
            ]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
            )
            returncode = await proc.wait()
            if proc.returncode:
                raise CalledProcessError(returncode=returncode, cmd=cmd)
            await zip_view.initialize_from_disk(temp_dir)