Module refinery.units.formats.archive.xt7z

Classes

class xt7z (*paths, list=False, join_path=False, drop_path=False, fuzzy=0, exact=False, regex=False, path=b'path', exclude=None, date=b'date', pwd=b'')

Extract files from a 7zip archive.

This unit extracts items with an associated virtual path from a container; each extracted item is emitted as a separate chunk with a corresponding meta variable named "path".

Positional arguments to xt7z are patterns to filter the extracted items. Use the -x flag to add an exclusion pattern. To extract all files with a foo or bar extension, but none that has the word "temp" in its path:

xt7z .foo .bar -x temp

To view only the paths of all chunks, use the listing switch:

emit data | ... | xt7z -l

Otherwise, extracted items are written to the standard output port and usually require a frame to properly process. In order to dump all extracted data to disk, the following pipeline can be used:

emit data | ... | xt7z [| dump extracted/{path} ]

The value {path} is a placeholder which is substituted by the virtual path of the extracted item. When using xt7z to unpack a file on disk, the following pattern can be useful:

ef pack.bin [| xt7z -j | d2p ]

The unit ef is also a path extractor. By specifying -j (or --join), the paths of extracted items are combined. Here, d2p is a shortcut for dump {path}. It deconflicts the joined paths with the local file system: If pack.bin contains items one.txt and two.txt, the following local file tree would be the result:

pack.bin
pack/one.txt
pack/two.txt

Finally, the -d (or --drop) switch can be used to not create (or alter) the path metadata at all, which is useful in cases where path metadata from a previous unit should be preserved.

Expand source code Browse git

class xt7z(ArchiveUnit, docs='{0}{p}{PathExtractorUnit}'):
    """
    Extract files from a 7zip archive.
    """

    def unpack(self, data: bytearray):
        for match in re.finditer(re.escape(SIGNATURE), data):
            start = match.start()
            if start != 0:
                self.log_info(F'found a header at offset 0x{start:X}, trying to extract from there.')
            try:
                yield from self._unpack_from(data, start)
            except SzCorruptArchive:
                continue
            else:
                break

    def _unpack_from(self, data: bytearray, zp: int = 0):
        mv = memoryview(data)
        chunk = mv[zp:]
        pwd = self.args.pwd

        def try_open(password: str | bytes | None) -> SzArchive:
            return SzArchive(chunk, password=password)

        archive: SzArchive | None = None
        extractors: list = []
        probed = None

        def prepare(archive: SzArchive, password):
            return [(info, info.extractor(password)) for info in archive.files if not info.is_dir]

        def checked(read):
            try:
                return read()
            except SzChecksumMismatch as mismatch:
                raise RefineryPartialResult(str(mismatch), mismatch.data) from mismatch

        if pwd:
            pwd = pwd.decode(self.codec)
            try:
                archive = try_open(pwd)
            except SzCorruptArchive:
                raise ValueError('corrupt archive; the password is likely invalid.')
            extractors = prepare(archive, pwd)
        else:
            def passwords():
                yield None
                yield from self.CommonPasswords
            unlocked = None
            for pwd in passwords():
                if pwd is None:
                    self.log_debug('trying empty password')
                else:
                    self.log_debug(F'trying password: {pwd}')
                trial = None
                try:
                    trial = try_open(pwd)
                    extractors = prepare(trial, pwd)
                    if pwd is None:
                        unlocked = trial, extractors
                    probed = next(((k, read()) for k, (info, read) in enumerate(extractors) if info.encrypted), None)
                    problem = False
                except SzUnsupportedMethod as E:
                    raise ValueError(str(E))
                except SzCannotUnpack:
                    problem = True
                except Exception:
                    if pwd is None:
                        raise
                    problem = True
                if not problem:
                    archive = trial
                    break
            else:
                if unlocked is None:
                    raise ValueError('a password is required and none of the default passwords worked.')
                self.log_info('none of the default passwords worked; extracting the files that are not encrypted')
                archive, extractors = unlocked
                probed = None

        assert archive is not None

        for k, (info, extract) in enumerate(extractors):
            yield self._pack(
                info.name,
                info.mtime or info.ctime,
                probed[1] if probed is not None and probed[0] == k else lambda read=extract: checked(read),
                checksum=info.crc,
                uncompressed=info.size,
            )

    @classmethod
    def handles(cls, data) -> bool | None:
        if data[:6] == SIGNATURE:
            return True
        if not is_likely_pe(data):
            return None
        offset = get_pe_size(data)
        memory = memoryview(data)
        memory = memory[offset:]
        if memory[:10] == B';!@Install' and buffer_offset(memory, SIGNATURE, 0, 0x1000) > 0:
            return True

Ancestors

• ArchiveUnit
• PathExtractorUnit
• Unit
• UnitBase
• Entry

Subclasses

• xt7z

Class variables

var reverse

The type of the None singleton.

Methods

def unpack(self, data)

Expand source code Browse git

def unpack(self, data: bytearray):
    for match in re.finditer(re.escape(SIGNATURE), data):
        start = match.start()
        if start != 0:
            self.log_info(F'found a header at offset 0x{start:X}, trying to extract from there.')
        try:
            yield from self._unpack_from(data, start)
        except SzCorruptArchive:
            continue
        else:
            break

Inherited members

• ArchiveUnit:
  • CommonPasswords
  • CustomJoinBehaviour
  • CustomPathSeparator
  • FilterEverything
  • Requires
  • act
  • assemble
  • carved
  • close
  • codec
  • console
  • filter
  • finish
  • handles
  • is_quiet
  • is_reversible
  • isatty
  • labelled
  • leniency
  • log_always
  • log_debug
  • log_detach
  • log_fail
  • log_info
  • log_level
  • log_warn
  • logger
  • name
  • nozzle
  • optional_dependencies
  • process
  • read
  • read1
  • required_dependencies
  • reset
  • run
  • source
  • superinit