Module refinery.units.formats.archive.xtzip

Classes

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

Extract files from a Zip archive. This unit is a path extractor which extracts data from a hierarchical structure. Each extracted item is emitted as a separate chunk and has attached to it a meta variable that contains its path within the source structure. The positional arguments to the command are patterns that can be used to filter the extracted items by their path. To view only the paths of all chunks, use the listing switch:

emit something | xtzip --list

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 something | xtzip [| dump {path} ]

Expand source code Browse git

class xtzip(ArchiveUnit, docs='{0}{s}{PathExtractorUnit}'):
    """
    Extract files from a Zip archive.
    """
    @ArchiveUnit.Requires('chardet', ['default', 'extended'])
    def _chardet():
        import chardet
        return chardet

    @ArchiveUnit.Requires('pyzipper', ['arc', 'default', 'extended'])
    def _pyzipper():
        import pyzipper
        return pyzipper

    @classmethod
    def _carver(cls):
        return carve_zip

    def unpack(self, data: bytearray):
        from zipfile import BadZipFile, ZipFile, ZipInfo

        def password_invalid(password: bytes | None):
            nonlocal archive, fallback
            if password:
                archive.setpassword(password)
            try:
                archive.testzip()
                files = (t for t in archive.infolist() if t.filename and not t.is_dir())
                files = sorted(files, key=lambda info: info.file_size)
                for info in files:
                    self.log_debug('testing password against:', info.filename)
                    try:
                        with archive.open(info.filename, "r") as test:
                            while test.read(1024):
                                pass
                    except BadZipFile:
                        continue
                    else:
                        break
            except NotImplementedError:
                if fallback:
                    raise
                self.log_debug('compression method unsupported, switching to pyzipper')
                archive = self._pyzipper.AESZipFile(MemoryFile(data))
                fallback = True
                return password_invalid(password)
            except RuntimeError as E:
                if 'password' not in str(E):
                    raise
                return True
            else:
                if password:
                    self.log_debug('using password:', password)
                return False

        password = bytes(self.args.pwd)
        fallback = False
        archive = ZipFile(MemoryFile(data))
        passwords = [password]

        if not password:
            passwords.extend(p.encode(self.codec) for p in self._COMMON_PASSWORDS)
        for p in passwords:
            if not password_invalid(p):
                break

        for info in archive.infolist():
            def xt(archive: ZipFile = archive, info: ZipInfo = info, data=memoryview(data)):
                try:
                    return archive.read(info.filename)
                except RuntimeError as E:
                    if 'password' not in str(E):
                        raise
                    msg = 'invalid password; use -L to extract raw encrypted data'
                    rec = _FileRecord(data[info.header_offset:])
                    raise RefineryPartialResult(msg, rec.data) from E

            if info.filename:
                if info.is_dir():
                    continue

            # courtesy of https://stackoverflow.com/a/37773438/9130824
            filename = info.filename
            if info.flag_bits & ZIP_FILENAME_UTF8_FLAG == 0:
                filename_bytes = filename.encode('437')
                try:
                    guessed_encoding = self._chardet.detect(filename_bytes)['encoding']
                except ImportError:
                    guessed_encoding = None
                guessed_encoding = guessed_encoding or 'cp1252'
                filename = filename_bytes.decode(guessed_encoding, 'replace')

            try:
                date = datetime(*info.date_time)
            except Exception as e:
                self.log_info(F'{e!s} - unable to determine date from tuple {info.date_time} for: {filename}')
                date = None

            yield self._pack(filename, date, xt)

    @classmethod
    def handles(cls, data: bytearray):
        return data.rfind(ZipEndOfCentralDirectory.SIGNATURE) > 0 and data[:2] == B'PK'

Ancestors

• ArchiveUnit
• PathExtractorUnit
• Unit
• UnitBase
• Entry

Subclasses

• xtzip

Class variables

var required_dependencies

var console

var reverse

var optional_dependencies

Methods

def unpack(self, data)

Expand source code Browse git

def unpack(self, data: bytearray):
    from zipfile import BadZipFile, ZipFile, ZipInfo

    def password_invalid(password: bytes | None):
        nonlocal archive, fallback
        if password:
            archive.setpassword(password)
        try:
            archive.testzip()
            files = (t for t in archive.infolist() if t.filename and not t.is_dir())
            files = sorted(files, key=lambda info: info.file_size)
            for info in files:
                self.log_debug('testing password against:', info.filename)
                try:
                    with archive.open(info.filename, "r") as test:
                        while test.read(1024):
                            pass
                except BadZipFile:
                    continue
                else:
                    break
        except NotImplementedError:
            if fallback:
                raise
            self.log_debug('compression method unsupported, switching to pyzipper')
            archive = self._pyzipper.AESZipFile(MemoryFile(data))
            fallback = True
            return password_invalid(password)
        except RuntimeError as E:
            if 'password' not in str(E):
                raise
            return True
        else:
            if password:
                self.log_debug('using password:', password)
            return False

    password = bytes(self.args.pwd)
    fallback = False
    archive = ZipFile(MemoryFile(data))
    passwords = [password]

    if not password:
        passwords.extend(p.encode(self.codec) for p in self._COMMON_PASSWORDS)
    for p in passwords:
        if not password_invalid(p):
            break

    for info in archive.infolist():
        def xt(archive: ZipFile = archive, info: ZipInfo = info, data=memoryview(data)):
            try:
                return archive.read(info.filename)
            except RuntimeError as E:
                if 'password' not in str(E):
                    raise
                msg = 'invalid password; use -L to extract raw encrypted data'
                rec = _FileRecord(data[info.header_offset:])
                raise RefineryPartialResult(msg, rec.data) from E

        if info.filename:
            if info.is_dir():
                continue

        # courtesy of https://stackoverflow.com/a/37773438/9130824
        filename = info.filename
        if info.flag_bits & ZIP_FILENAME_UTF8_FLAG == 0:
            filename_bytes = filename.encode('437')
            try:
                guessed_encoding = self._chardet.detect(filename_bytes)['encoding']
            except ImportError:
                guessed_encoding = None
            guessed_encoding = guessed_encoding or 'cp1252'
            filename = filename_bytes.decode(guessed_encoding, 'replace')

        try:
            date = datetime(*info.date_time)
        except Exception as e:
            self.log_info(F'{e!s} - unable to determine date from tuple {info.date_time} for: {filename}')
            date = None

        yield self._pack(filename, date, xt)

Inherited members

• ArchiveUnit:
  • CustomPathSeparator
  • Requires
  • assemble
  • filter
  • finish
  • handles
  • is_quiet
  • labelled
  • leniency
  • log_always
  • log_debug
  • log_detach
  • log_fail
  • log_info
  • log_level
  • log_warn
  • nozzle
  • process
  • read
  • read1
  • run
  • source
  • superinit