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
    list_files_output = subprocess.run(
        ["find", ".", "-print"],
        check=True,
        capture_output=True,
        cwd=temp_flush_dir,
    )
    cpio_pack_output = subprocess.run(
        ["cpio", "-o", f"--format={cpio_format}"],
        check=True,
        capture_output=True,
        cwd=temp_flush_dir,
        input=list_files_output.stdout,
    )
    new_data = cpio_pack_output.stdout
    # 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.TemporaryDirectory() as temp_flush_dir:
        command = ["cpio", "-id"]
        subprocess.run(
            command,
            check=True,
            capture_output=True,
            cwd=temp_flush_dir,
            input=resource_data,
        )
        await cpio_v.initialize_from_disk(temp_flush_dir)