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	ZipPackerConfig	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: ZipPackerConfig = None) -> None:
    if config is None:
        config = ZipPackerConfig()
    if not 0 <= config.compression_level <= 9:
        raise ValueError("compression_level must be an integer from 0-9")

    zip_view: ZipArchive = await resource.view_as(ZipArchive)
    flush_dir = await zip_view.flush_to_disk()

    with tempfile.NamedTemporaryFile(suffix=".zip", delete_on_close=False) as temp_archive:
        temp_archive.close()
        os.unlink(
            temp_archive.name
        )  # zip fails if the output path exists but isn't a valid zip
        cmd = [
            "zip",
            f"-{config.compression_level}",
            "-r",
            temp_archive.name,
            ".",
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            cwd=flush_dir,
        )
        returncode = await proc.wait()
        if proc.returncode:
            raise CalledProcessError(returncode=returncode, cmd=cmd)

        with open(temp_archive.name, "rb") as fh:
            resource.queue_patch(Range(0, await resource.get_data_length()), fh.read())

ZipPackerConfig (ComponentConfig) dataclass

Configuration for ZipPacker.

Parameters:

Name	Type	Description	Default
compression_level		ZIP compression level from 0 (store, no compression) to 9 (maximum compression). Higher levels produce smaller archives at the cost of longer packing times. Defaults to 6, which balances compression ratio and speed.	required

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	ComponentConfig	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: ComponentConfig = None) -> 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)