cpio.py

ofrak.core.cpio

CpioArchiveType (Enum)

CPIO has several unrelated, incompatible variants. They're described in the man page: https://linux.die.net/man/1/cpio

CpioFilesystem (GenericBinary, FilesystemRoot) dataclass

Filesystem stored in a CPIO archive.

CpioFilesystemAnalyzer (Analyzer)

Extracts CPIO archive metadata including archive format type (newc, crc, old binary, etc.), number of files, total archive size, and format-specific parameters. Use before modifying CPIO archives to preserve the correct format, ensure compatibility with target systems (some bootloaders only support specific formats), or understand archive characteristics. Essential for maintaining format compatibility during repacking.

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
CpioFilesystem	The analysis results

Source code in ofrak/core/cpio.py

async def analyze(self, resource: Resource, config=None) -> CpioFilesystem:
    _magic = await resource.analyze(Magic)
    magic_description = _magic.descriptor
    if magic_description.startswith("ASCII cpio archive (SVR4 with no CRC)"):
        archive_type = CpioArchiveType.NEW_ASCII
    elif magic_description.startswith("ASCII cpio archive (pre-SVR4 or odc)"):
        archive_type = CpioArchiveType.OLD_ASCII
    elif magic_description.startswith("ASCII cpio archive (SVR4 with CRC)"):
        archive_type = CpioArchiveType.CRC_ASCII
    elif magic_description.startswith("cpio archive"):
        archive_type = CpioArchiveType.BINARY
    else:
        raise NotImplementedError(
            f"Please add support for CPIO archive type {magic_description}"
        )

    return CpioFilesystem(archive_type)

CpioPacker (Packer)

Packages files into a CPIO archive format suitable for use as Linux initial ramdisk (initramfs) or embedded firmware packaging. Use after modifying extracted CPIO contents to recreate initramfs images or firmware archives. Critical for modifying Linux boot process and embedded system initialization.

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

async def pack(self, resource: Resource, config=None):
    cpio_v: CpioFilesystem = await resource.view_as(CpioFilesystem)
    temp_flush_dir = await cpio_v.flush_to_disk()
    cpio_format = cpio_v.archive_type.value
    list_files_cmd = [
        "find",
        "-print",
    ]
    list_files_proc = await asyncio.create_subprocess_exec(
        *list_files_cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=temp_flush_dir,
    )
    list_files_list, stderr = await list_files_proc.communicate()
    if list_files_proc.returncode:
        raise CalledProcessError(returncode=list_files_proc.returncode, cmd=list_files_cmd)

    cpio_pack_cmd = [
        "cpio",
        "-o",
        f"--format={cpio_format}",
    ]
    cpio_pack_proc = await asyncio.create_subprocess_exec(
        *cpio_pack_cmd,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=temp_flush_dir,
    )
    cpio_pack_output, stderr = await cpio_pack_proc.communicate(input=list_files_list)
    if cpio_pack_proc.returncode:
        raise CalledProcessError(returncode=cpio_pack_proc.returncode, cmd=cpio_pack_cmd)
    # Passing in the original range effectively replaces the original data with the new data
    resource.queue_patch(Range(0, await resource.get_data_length()), cpio_pack_output)

CpioUnpacker (Unpacker)

Extracts files and directories from CPIO archive formats, which are commonly used in Linux initial ramdisk (initramfs) images and some embedded system firmware packages. CPIO archives can contain regular files, directories, symbolic links, and special device files with preserved permissions and ownership. Use when analyzing Linux boot processes (initramfs/initrd), embedded firmware that uses CPIO for packaging, or any CPIO archive. Essential for examining and modifying initramfs filesystems.

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

async def unpack(self, resource: Resource, config=None):
    cpio_v = await resource.view_as(CpioFilesystem)
    resource_data = await cpio_v.resource.get_data()
    with tempfile.TemporaryDirectory() as temp_flush_dir:
        cmd = [
            "cpio",
            "-id",
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=temp_flush_dir,
        )
        await proc.communicate(input=resource_data)
        if proc.returncode:
            raise CalledProcessError(returncode=proc.returncode, cmd=cmd)
        await cpio_v.initialize_from_disk(temp_flush_dir)