Skip to content

cpio.py

ofrak_components.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)

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_components/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)

Pack files into a CPIO archive.

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_components/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
    with tempfile.NamedTemporaryFile(suffix=".cpio", mode="rb") as temp:
        # Use subshell to handle relative paths and avoid changing directories back and forth
        command = f"(cd {temp_flush_dir} && find . -print | cpio -o --format={cpio_format} > {temp.name})"
        try:
            subprocess.run(command, check=True, capture_output=True, shell=True)
        except CalledProcessError as error:
            raise PackerError(format_called_process_error(error))
        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)

CpioUnpacker (Unpacker)

Unpack a CPIO archive.

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_components/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.NamedTemporaryFile() as temp_file:
        temp_file.write(resource_data)
        temp_file.flush()
        with tempfile.TemporaryDirectory() as temp_flush_dir:
            # Use subshell to handle relative paths and avoid changing directories back and
            # forth
            command = f"(cd {temp_flush_dir} && cpio -id < {temp_file.name})"
            try:
                subprocess.run(command, check=True, capture_output=True, shell=True)
            except subprocess.CalledProcessError as error:
                raise UnpackerError(format_called_process_error(error))
            await cpio_v.initialize_from_disk(temp_flush_dir)