Module refinery.units.formats.archive.xtrar

Classes

class xtrar (*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 RAR 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 xtrar 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:

xtrar .foo .bar -x temp

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

emit data | ... | xtrar -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 | ... | xtrar [| dump extracted/{path} ]

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

ef pack.bin [| xtrar -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 xtrar(ArchiveUnit, docs='{0}{p}{PathExtractorUnit}'):
    """
    Extract files from a RAR archive.
    """
    @staticmethod
    def _find_rar_file(data: buf):
        view = memoryview(data)
        if detect_format(view) is not None:
            return view
        from refinery.lib.id import get_pe_type
        if get_pe_type(data):
            from refinery.units.formats.pe import get_pe_size
            offset = get_pe_size(data)
            overlay = view[offset:]
            if detect_format(overlay) is not None:
                return overlay
        for signature in (
            RAR_HEADER_V50,
            RAR_HEADER_V15,
            RAR_HEADER_V14,
        ):
            offset = buffer_offset(view, signature)
            if offset >= 0:
                return view[offset:]

    def unpack(self, data):
        if (view := self._find_rar_file(data)) is None:
            raise ValueError('Input does not appear to be a RAR file.')
        password = self.args.pwd
        if not password:
            password = None
        elif not isinstance(password, str):
            password = codecs.decode(password, self.codec)
        rar = RarFile(view, password=password)
        if rar.is_encrypted and not password and not rar.entries:
            raise RarMissingPassword
        files = [entry for entry in rar.entries if not entry.is_dir]
        extractors = [rar.extractor(entry, password) for entry in files]
        probed = None
        if not self.args.list and not password:
            encrypted = next((k for k, entry in enumerate(files) if entry.is_encrypted and entry.size != 0), None)
            if encrypted is not None:
                for pwd in self.CommonPasswords:
                    trial = [rar.extractor(entry, pwd) for entry in files]
                    try:
                        data = trial[encrypted]()
                    except (
                        RarMissingPassword,
                        RarInvalidPassword,
                        RarInvalidChecksum,
                        RarCorruptArchive,
                    ):
                        continue
                    except Exception:
                        break
                    else:
                        self.log_info(F'using password: {pwd}')
                        extractors = trial
                        probed = encrypted, data
                        break
        for k, (entry, read) in enumerate(zip(files, extractors)):
            if probed is not None and probed[0] == k:
                yield self._pack(entry.name, entry.date, probed[1])
                continue

            def extract(read=read, entry=entry):
                try:
                    return read()
                except RarInvalidChecksum as check:
                    raise RefineryPartialResult(str(check), check.data) from check
                except RarCorruptArchive as exc:
                    if exc.partial is None:
                        raise
                    raise RefineryPartialResult(str(exc), exc.partial) from exc
                except RarInvalidPassword as E:
                    raise ValueError(F'invalid password: {entry.name}') from E
                except RarMissingPassword as E:
                    raise ValueError(F'missing password: {entry.name}') from E
            yield self._pack(entry.name, entry.date, extract)

    @classmethod
    def handles(cls, data) -> bool:
        return cls._find_rar_file(data) is not None

Ancestors

• ArchiveUnit
• PathExtractorUnit
• Unit
• UnitBase
• Entry

Subclasses

• xtrar

Class variables

var reverse

The type of the None singleton.

Methods

def unpack(self, data)

Expand source code Browse git

def unpack(self, data):
    if (view := self._find_rar_file(data)) is None:
        raise ValueError('Input does not appear to be a RAR file.')
    password = self.args.pwd
    if not password:
        password = None
    elif not isinstance(password, str):
        password = codecs.decode(password, self.codec)
    rar = RarFile(view, password=password)
    if rar.is_encrypted and not password and not rar.entries:
        raise RarMissingPassword
    files = [entry for entry in rar.entries if not entry.is_dir]
    extractors = [rar.extractor(entry, password) for entry in files]
    probed = None
    if not self.args.list and not password:
        encrypted = next((k for k, entry in enumerate(files) if entry.is_encrypted and entry.size != 0), None)
        if encrypted is not None:
            for pwd in self.CommonPasswords:
                trial = [rar.extractor(entry, pwd) for entry in files]
                try:
                    data = trial[encrypted]()
                except (
                    RarMissingPassword,
                    RarInvalidPassword,
                    RarInvalidChecksum,
                    RarCorruptArchive,
                ):
                    continue
                except Exception:
                    break
                else:
                    self.log_info(F'using password: {pwd}')
                    extractors = trial
                    probed = encrypted, data
                    break
    for k, (entry, read) in enumerate(zip(files, extractors)):
        if probed is not None and probed[0] == k:
            yield self._pack(entry.name, entry.date, probed[1])
            continue

        def extract(read=read, entry=entry):
            try:
                return read()
            except RarInvalidChecksum as check:
                raise RefineryPartialResult(str(check), check.data) from check
            except RarCorruptArchive as exc:
                if exc.partial is None:
                    raise
                raise RefineryPartialResult(str(exc), exc.partial) from exc
            except RarInvalidPassword as E:
                raise ValueError(F'invalid password: {entry.name}') from E
            except RarMissingPassword as E:
                raise ValueError(F'missing password: {entry.name}') from E
        yield self._pack(entry.name, entry.date, extract)

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