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)

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)

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

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