monkee.ch/SickGear
1
0
Fork 0
mirror of https://github.com/SickGear/SickGear.git synced 2024-11-26 22:53:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2
SickGear/lib/hachoir/parser/archive/sevenzip.py

827 lines
27 KiB
Python
Raw Normal View History

Change bump to major version 3.xx to signal that this branch supports Python3+ only. Initial SickGear for Python 3.
2023-01-12 01:04:47 +00:00
"""
7zip file parser
Informations:
- File 7zformat.txt of 7-zip SDK:
http://www.7-zip.org/sdk.html
Author: Olivier SCHWAB
Creation date: 6 december 2006
Updated by: Robert Xiao
Date: February 26 2011
"""
from hachoir.parser import Parser
from hachoir.field import (Field, FieldSet, ParserError, CString,
Enum, Bit, Bits, UInt8, UInt32, UInt64,
Bytes, RawBytes, TimestampWin64)
from hachoir.stream import StringInputStream
from hachoir.core.endian import LITTLE_ENDIAN
from hachoir.core.text_handler import textHandler, hexadecimal
from hachoir.core.tools import alignValue, humanFilesize, makePrintable
from hachoir.parser.common.msdos import MSDOSFileAttr32
try:
from pylzma import decompress as lzmadecompress
has_lzma = True
except ImportError:
has_lzma = False
class SZUInt64(Field):
"""
Variable length UInt64, where the first byte gives both the number of bytes
needed and the upper byte value.
"""
def __init__(self, parent, name, max_size=None, description=None):
Field.__init__(self, parent, name, size=8, description=description)
value = 0
addr = self.absolute_address
mask = 0x80
firstByte = parent.stream.readBits(addr, 8, LITTLE_ENDIAN)
for i in range(8):
addr += 8
if not (firstByte & mask):
value += ((firstByte & (mask - 1)) << (8 * i))
break
value |= (parent.stream.readBits(
addr, 8, LITTLE_ENDIAN) << (8 * i))
mask >>= 1
self._size += 8
self.createValue = lambda: value
kEnd = 0x00
kHeader = 0x01
kArchiveProperties = 0x02
kAdditionalStreamsInfo = 0x03
kMainStreamsInfo = 0x04
kFilesInfo = 0x05
kPackInfo = 0x06
kUnPackInfo = 0x07
kSubStreamsInfo = 0x08
kSize = 0x09
kCRC = 0x0A
kFolder = 0x0B
kCodersUnPackSize = 0x0C
kNumUnPackStream = 0x0D
kEmptyStream = 0x0E
kEmptyFile = 0x0F
kAnti = 0x10
kName = 0x11
kCreationTime = 0x12
kLastAccessTime = 0x13
kLastWriteTime = 0x14
kWinAttributes = 0x15
kComment = 0x16
kEncodedHeader = 0x17
kStartPos = 0x18
kDummy = 0x19
PROP_IDS = {
kEnd: 'kEnd',
kHeader: 'kHeader',
kArchiveProperties: 'kArchiveProperties',
kAdditionalStreamsInfo: 'kAdditionalStreamsInfo',
kMainStreamsInfo: 'kMainStreamsInfo',
kFilesInfo: 'kFilesInfo',
kPackInfo: 'kPackInfo',
kUnPackInfo: 'kUnPackInfo',
kSubStreamsInfo: 'kSubStreamsInfo',
kSize: 'kSize',
kCRC: 'kCRC',
kFolder: 'kFolder',
kCodersUnPackSize: 'kCodersUnPackSize',
kNumUnPackStream: 'kNumUnPackStream',
kEmptyStream: 'kEmptyStream',
kEmptyFile: 'kEmptyFile',
kAnti: 'kAnti',
kName: 'kName',
kCreationTime: 'kCreationTime',
kLastAccessTime: 'kLastAccessTime',
kLastWriteTime: 'kLastWriteTime',
kWinAttributes: 'kWinAttributes',
kComment: 'kComment',
kEncodedHeader: 'kEncodedHeader',
kStartPos: 'kStartPos',
kDummy: 'kDummy',
}
PROP_DESC = {
kEnd: 'End-of-header marker',
kHeader: 'Archive header',
kArchiveProperties: 'Archive properties',
kAdditionalStreamsInfo: 'AdditionalStreamsInfo',
kMainStreamsInfo: 'MainStreamsInfo',
kFilesInfo: 'FilesInfo',
kPackInfo: 'PackInfo',
kUnPackInfo: 'UnPackInfo',
kSubStreamsInfo: 'SubStreamsInfo',
kSize: 'Size',
kCRC: 'CRC',
kFolder: 'Folder',
kCodersUnPackSize: 'CodersUnPackSize',
kNumUnPackStream: 'NumUnPackStream',
kEmptyStream: 'EmptyStream',
kEmptyFile: 'EmptyFile',
kAnti: 'Anti',
kName: 'Name',
kCreationTime: 'CreationTime',
kLastAccessTime: 'LastAccessTime',
kLastWriteTime: 'LastWriteTime',
kWinAttributes: 'WinAttributes',
kComment: 'Comment',
kEncodedHeader: 'Encoded archive header',
kStartPos: 'Unknown',
kDummy: 'Dummy entry',
}
def ReadNextByte(self):
return self.stream.readBits(self.absolute_address + self.current_size, 8, self.endian)
def PropID(self, name):
return Enum(UInt8(self, name), PROP_IDS)
class SevenZipBitVector(FieldSet):
def __init__(self, parent, name, num, has_all_byte=False, **args):
FieldSet.__init__(self, parent, name, **args)
self.has_all_byte = has_all_byte
self.num = num
def createFields(self):
if self.has_all_byte:
yield Enum(UInt8(self, "all_defined"), {0: 'False', 1: 'True'})
if self['all_defined'].value:
return
nbytes = alignValue(self.num, 8) // 8
ctr = 0
for i in range(nbytes):
for j in reversed(range(8)):
yield Bit(self, "bit[%d]" % (ctr + j))
ctr += 8
def isAllDefined(self):
return self.has_all_byte and self['all_defined'].value
def isDefined(self, index):
if self.isAllDefined():
return True
return self['bit[%d]' % index].value
def createValue(self):
if self.isAllDefined():
return list(range(self.num))
return [i for i in range(self.num) if self['bit[%d]' % i].value]
def createDisplay(self):
if self.isAllDefined():
return 'all'
return ','.join(str(i) for i in self.value)
class ArchiveProperty(FieldSet):
def createFields(self):
yield PropID(self, "id")
size = SZUInt64(self, "size")
yield size
if size.value:
yield RawBytes(self, "data", size.value)
def createDescription(self):
return self['id'].display
class ArchiveProperties(FieldSet):
def createFields(self):
yield PropID(self, "id")
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end_marker")
break
yield ArchiveProperty(self, "prop[]")
class Digests(FieldSet):
def __init__(self, parent, name, num_digests, digest_desc=None, desc=None):
FieldSet.__init__(self, parent, name, desc)
self.num_digests = num_digests
if digest_desc is None:
self.digest_desc = ['stream %d' % i for i in range(num_digests)]
else:
self.digest_desc = digest_desc
def createFields(self):
yield PropID(self, "id")
definearr = SevenZipBitVector(
self, "defined", self.num_digests, has_all_byte=True)
yield definearr
for index in definearr.value:
yield textHandler(UInt32(self, "digest[]",
"Digest for %s" % self.digest_desc[index]), hexadecimal)
class PackInfo(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "pack_pos", "File offset to the packed data")
num = SZUInt64(self, "num_pack_streams", "Number of packed streams")
yield num
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end_marker")
break
elif uid == kSize:
yield PropID(self, "size_marker")
for index in range(num.value):
yield SZUInt64(self, "pack_size[]")
elif uid == kCRC:
yield Digests(self, "digests", num.value)
else:
raise ParserError("Unexpected ID (%i)" % uid)
METHODS = {
b"\0": "Copy",
b"\3": "Delta",
b"\4": "x86_BCJ",
b"\5": "PowerPC",
b"\6": "IA64",
b"\7": "ARM_LE",
b"\x08": "ARMT_LE", # thumb
b"\x09": "SPARC",
b"\x21": "LZMA2",
b"\2\3\2": "Common-Swap-2",
b"\2\3\4": "Common-Swap-4",
b"\3\1\1": "7z-LZMA",
b"\3\3\1\3": "7z-Branch-x86-BCJ",
b"\3\3\1\x1b": "7z-Branch-x86-BCJ2",
b"\3\3\2\5": "7z-Branch-PowerPC-BE",
b"\3\3\3\1": "7z-Branch-Alpha-LE",
b"\3\3\4\1": "7z-Branch-IA64-LE",
b"\3\3\5\1": "7z-Branch-ARM-LE",
b"\3\3\6\5": "7z-Branch-M68-BE",
b"\3\3\7\1": "7z-Branch-ARMT-LE",
b"\3\3\x08\5": "7z-Branch-SPARC-BE",
b"\3\4\1": "7z-PPMD",
b"\3\x7f\1": "7z-Experimental",
b"\4\0": "Reserved",
b"\4\1\0": "Zip-Copy",
b"\4\1\1": "Zip-Shrink",
b"\4\1\6": "Zip-Implode",
b"\4\1\x08": "Zip-Deflate",
b"\4\1\x09": "Zip-Deflate64",
b"\4\1\x10": "Zip-BZip2",
b"\4\1\x14": "Zip-LZMA",
b"\4\1\x60": "Zip-JPEG",
b"\4\1\x61": "Zip-WavPack",
b"\4\1\x62": "Zip-PPMD",
b"\4\1\x63": "Zip-wzAES",
b"\4\2\2": "BZip2",
b"\4\3\1": "RAR-15",
b"\4\3\2": "RAR-20",
b"\4\3\3": "RAR-29",
b"\4\4\1": "Arj3",
b"\4\4\2": "Arj4",
b"\4\5": "Z",
b"\4\6": "LZH",
b"\4\7": "7z-Reserved",
b"\4\x08": "CAB",
b"\4\x09\1": "NSIS-Deflate",
b"\4\x09\2": "NSIS-BZip2",
b"\6\0": "Crypto-Reserved",
b"\6\1\x00": "Crypto-AES128-ECB",
b"\6\1\x01": "Crypto-AES128-CBC",
b"\6\1\x02": "Crypto-AES128-CFB",
b"\6\1\x03": "Crypto-AES128-OFB",
b"\6\1\x04": "Crypto-AES128-CTR",
b"\6\1\x40": "Crypto-AES192-ECB",
b"\6\1\x41": "Crypto-AES192-CBC",
b"\6\1\x42": "Crypto-AES192-CFB",
b"\6\1\x43": "Crypto-AES192-OFB",
b"\6\1\x44": "Crypto-AES192-CTR",
b"\6\1\x80": "Crypto-AES256-ECB",
b"\6\1\x81": "Crypto-AES256-CBC",
b"\6\1\x82": "Crypto-AES256-CFB",
b"\6\1\x83": "Crypto-AES256-OFB",
b"\6\1\x84": "Crypto-AES256-CTR",
b"\6\1\xc0": "Crypto-AES-ECB",
b"\6\1\xc1": "Crypto-AES-CBC",
b"\6\1\xc2": "Crypto-AES-CFB",
b"\6\1\xc3": "Crypto-AES-OFB",
b"\6\1\xc4": "Crypto-AES-CTR",
b"\6\7": "Crypto-Reserved",
b"\6\x0f": "Crypto-Reserved",
b"\6\xf0": "Crypto-Misc",
b"\6\xf1\1\1": "Crypto-Zip",
b"\6\xf1\3\2": "Crypto-RAR-Unknown",
b"\6\xf1\3\3": "Crypto-RAR-29", # AES128
b"\6\xf1\7\1": "Crypto-7z", # AES256
b"\7\0": "Hash-None",
b"\7\1": "Hash-CRC",
b"\7\2": "Hash-SHA1",
b"\7\3": "Hash-SHA256",
b"\7\4": "Hash-SHA384",
b"\7\5": "Hash-SHA512",
b"\7\xf0": "Hash-Misc",
b"\7\xf1\3\3": "Hash-RAR-29", # modified SHA1
b"\7\xf1\7\1": "Hash-7z", # SHA256
}
class Coder(FieldSet):
def createFields(self):
yield Bits(self, "id_size", 4)
yield Bit(self, "is_not_simple", "If unset, stream setup is simple")
yield Bit(self, "has_attribs", "Are there compression properties attached?")
yield Bit(self, "unused[]")
yield Bit(self, "is_not_last_method", "Are there more methods after this one in the alternative method list?")
size = self['id_size'].value
if size > 0:
yield Enum(RawBytes(self, "id", size), METHODS)
if self['is_not_simple'].value:
yield SZUInt64(self, "num_stream_in")
yield SZUInt64(self, "num_stream_out")
self.info("Streams: IN=%u OUT=%u" %
(self["num_stream_in"].value, self["num_stream_out"].value))
if self['has_attribs'].value:
size = SZUInt64(self, "properties_size")
yield size
yield RawBytes(self, "properties", size.value)
def _get_num_streams(self, direction):
if self['is_not_simple'].value:
return self['num_stream_%s' % direction].value
return 1
in_streams = property(lambda self: self._get_num_streams('in'))
out_streams = property(lambda self: self._get_num_streams('out'))
class CoderList(FieldSet):
def createFields(self):
while not self.eof:
field = Coder(self, "coder[]")
yield field
if not field['is_not_last_method'].value:
break
class BindPairInfo(FieldSet):
def createFields(self):
# 64 bits values then cast to 32 in fact
yield SZUInt64(self, "in_index")
yield SZUInt64(self, "out_index")
self.info("Indexes: IN=%u OUT=%u" %
(self["in_index"].value, self["out_index"].value))
class Folder(FieldSet):
def createFields(self):
yield SZUInt64(self, "num_coders")
num = self["num_coders"].value
self.info("Folder: %u codecs" % num)
in_streams = out_streams = 0
# Coder info
for index in range(num):
ci = CoderList(self, "coders[]")
yield ci
in_streams += ci['coder[0]'].in_streams
out_streams += ci['coder[0]'].out_streams
self._in_streams = in_streams
self._out_streams = out_streams
# Bind pairs
self.info("out streams: %u" % out_streams)
for index in range(out_streams - 1):
yield BindPairInfo(self, "bind_pair[]")
# Packed streams
# @todo: Actually find mapping
packed_streams = in_streams - out_streams + 1
if packed_streams > 1:
for index in range(packed_streams):
yield SZUInt64(self, "pack_stream[]")
def _get_num_streams(self, direction):
list(self)
return getattr(self, '_' + direction + '_streams')
in_streams = property(lambda self: self._get_num_streams('in'))
out_streams = property(lambda self: self._get_num_streams('out'))
class UnpackInfo(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield PropID(self, "folder_marker")
assert self['folder_marker'].value == kFolder
yield SZUInt64(self, "num_folders")
# Get generic info
num = self["num_folders"].value
self.info("%u folders" % num)
yield UInt8(self, "is_external")
if self['is_external'].value:
yield SZUInt64(self, "folder_data_offset", "Offset to folder data within data stream")
else:
# Read folder items
for folder_index in range(num):
yield Folder(self, "folder[]")
yield PropID(self, "unpacksize_marker")
assert self['unpacksize_marker'].value == kCodersUnPackSize
for folder_index in range(num):
folder = self["folder[%u]" % folder_index]
for index in range(folder.out_streams):
yield SZUInt64(self, "unpack_size[%d][%d]" % (folder_index, index))
# Extract digests
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end_marker")
break
elif uid == kCRC:
yield Digests(self, "digests", num)
else:
raise ParserError("Unexpected ID (%i)" % uid)
class SubStreamInfo(FieldSet):
def createFields(self):
yield PropID(self, "id")
num_folders = self['../unpack_info/num_folders'].value
num_unpackstreams = [1] * num_folders
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end_marker")
break
elif uid == kNumUnPackStream:
yield PropID(self, "num_unpackstream_marker")
for i in range(num_folders):
field = SZUInt64(self, "num_unpackstreams[]")
yield field
num_unpackstreams[i] = field.value
elif uid == kSize:
yield PropID(self, "size_marker")
for i in range(num_folders):
# The last substream's size is the stream size minus the
# other substreams.
for j in range(num_unpackstreams[i] - 1):
yield SZUInt64(self, "unpack_size[%d][%d]" % (i, j))
elif uid == kCRC:
digests = []
for i in range(num_folders):
if num_unpackstreams[i] == 1 and 'digests' in self['../unpack_info']:
continue
for j in range(num_unpackstreams[i]):
digests.append('folder %i, stream %i' % (i, j))
yield Digests(self, "digests", len(digests), digests)
else:
raise ParserError("Unexpected ID (%i)" % uid)
class StreamsInfo(FieldSet):
def createFields(self):
yield PropID(self, "id")
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end")
break
elif uid == kPackInfo:
yield PackInfo(self, "pack_info", PROP_DESC[uid])
elif uid == kUnPackInfo:
yield UnpackInfo(self, "unpack_info", PROP_DESC[uid])
elif uid == kSubStreamsInfo:
yield SubStreamInfo(self, "substreams_info", PROP_DESC[uid])
else:
raise ParserError("Unexpected ID (%i)" % uid)
class EncodedHeader(StreamsInfo):
pass
class EmptyStreamProperty(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "size")
yield SevenZipBitVector(self, "vec", self['../num_files'].value)
def createValue(self):
return self['vec'].value
def createDisplay(self):
return self['vec'].display
class EmptyFileProperty(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "size")
empty_streams = self['../empty_streams/vec'].value
yield SevenZipBitVector(self, "vec", len(empty_streams))
def createValue(self):
empty_streams = self['../empty_streams/vec'].value
return [empty_streams[i] for i in self['vec'].value]
def createDisplay(self):
return ','.join(str(i) for i in self.value)
class FileTimeProperty(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "size")
definearr = SevenZipBitVector(
self, "defined", self['../num_files'].value, has_all_byte=True)
yield definearr
yield UInt8(self, "is_external")
if self['is_external'].value:
yield SZUInt64(self, "folder_data_offset", "Offset to folder data within data stream")
else:
for index in definearr.value:
yield TimestampWin64(self, "timestamp[%d]" % index)
class FileNames(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "size")
yield UInt8(self, "is_external")
if self['is_external'].value:
yield SZUInt64(self, "folder_data_offset", "Offset to folder data within data stream")
else:
for index in range(self['../num_files'].value):
yield CString(self, "name[%d]" % index, charset="UTF-16-LE")
class FileAttributes(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "size")
definearr = SevenZipBitVector(
self, "defined", self['../num_files'].value, has_all_byte=True)
yield definearr
yield UInt8(self, "is_external")
if self['is_external'].value:
yield SZUInt64(self, "folder_data_offset", "Offset to folder data within data stream")
else:
for index in definearr.value:
yield MSDOSFileAttr32(self, "attributes[%d]" % index)
class FilesInfo(FieldSet):
def createFields(self):
yield PropID(self, "id")
yield SZUInt64(self, "num_files")
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end_marker")
break
elif uid == kEmptyStream:
yield EmptyStreamProperty(self, "empty_streams")
elif uid == kEmptyFile:
yield EmptyFileProperty(self, "empty_files")
elif uid == kAnti:
yield EmptyFileProperty(self, "anti_files")
elif uid == kCreationTime:
yield FileTimeProperty(self, "creation_time")
elif uid == kLastAccessTime:
yield FileTimeProperty(self, "access_time")
elif uid == kLastWriteTime:
yield FileTimeProperty(self, "modified_time")
elif uid == kName:
yield FileNames(self, "filenames")
elif uid == kWinAttributes:
yield FileAttributes(self, "attributes")
elif uid == kDummy:
yield ArchiveProperty(self, "dummy[]")
else:
yield ArchiveProperty(self, "prop[]")
class Header(FieldSet):
def createFields(self):
yield PropID(self, "id")
while not self.eof:
uid = ReadNextByte(self)
if uid == kEnd:
yield PropID(self, "end")
break
elif uid == kArchiveProperties:
yield ArchiveProperties(self, "props", PROP_DESC[uid])
elif uid == kAdditionalStreamsInfo:
yield StreamsInfo(self, "additional_streams", PROP_DESC[uid])
elif uid == kMainStreamsInfo:
yield StreamsInfo(self, "main_streams", PROP_DESC[uid])
elif uid == kFilesInfo:
yield FilesInfo(self, "files_info", PROP_DESC[uid])
else:
raise ParserError("Unexpected ID %u" % uid)
class NextHeader(FieldSet):
def __init__(self, parent, name, desc="Next header"):
FieldSet.__init__(self, parent, name, desc)
self._size = 8 * self["/signature/start_hdr/next_hdr_size"].value
def createFields(self):
uid = ReadNextByte(self)
if uid == kHeader:
yield Header(self, "header", PROP_DESC[uid])
elif uid == kEncodedHeader:
yield EncodedHeader(self, "encoded_hdr", PROP_DESC[uid])
else:
raise ParserError("Unexpected ID %u" % uid)
class NextHeaderParser(Parser):
PARSER_TAGS = {
}
endian = LITTLE_ENDIAN
def createFields(self):
uid = ReadNextByte(self)
if uid == kHeader:
yield Header(self, "header", PROP_DESC[uid])
elif uid == kEncodedHeader:
yield EncodedHeader(self, "encoded_hdr", PROP_DESC[uid])
else:
raise ParserError("Unexpected ID %u" % uid)
def validate(self):
return True
class CompressedData(Bytes):
def __init__(self, parent, name, length, decompressor, description=None,
parser=None, filename=None, mime_type=None, parser_class=None):
if filename:
if not isinstance(filename, str):
filename = makePrintable(filename, "ISO-8859-1")
if not description:
description = 'File "%s" (%s)' % (
filename, humanFilesize(length))
Bytes.__init__(self, parent, name, length, description)
self.setupInputStream(decompressor, parser,
filename, mime_type, parser_class)
def setupInputStream(self, decompressor, parser, filename, mime_type, parser_class):
def createInputStream(cis, **args):
tags = args.setdefault("tags", [])
if parser_class:
tags.append(("class", parser_class))
if parser is not None:
tags.append(("id", parser.PARSER_TAGS["id"]))
if mime_type:
tags.append(("mime", mime_type))
if filename:
tags.append(("filename", filename))
return StringInputStream(decompressor(self.value), **args)
self.setSubIStream(createInputStream)
def get_header_decompressor(self):
unpack_info = self['/next_hdr/encoded_hdr/unpack_info']
assert unpack_info['num_folders'].value == 1
coder = unpack_info['folder[0]/coders[0]/coder[0]']
method = METHODS[coder['id'].value]
if method == 'Copy':
return lambda data: data
elif method == '7z-LZMA' and has_lzma:
props = coder['properties'].value
length = unpack_info['unpack_size[0][0]'].value
return lambda data: lzmadecompress(props + data, maxlength=length)
def get_header_field(self, name, size, description=None):
decompressor = get_header_decompressor(self)
if decompressor is None:
return RawBytes(self, name, size, description=description)
return CompressedData(self, name, size, decompressor, description=description, parser_class=NextHeaderParser)
class Body(FieldSet):
def __init__(self, parent, name, desc="Body data"):
FieldSet.__init__(self, parent, name, desc)
self._size = 8 * self["/signature/start_hdr/next_hdr_offset"].value
def createFields(self):
if "encoded_hdr" in self["/next_hdr"]:
pack_size = sum([s.value for s in self.array(
"/next_hdr/encoded_hdr/pack_info/pack_size")])
body_size = self["/next_hdr/encoded_hdr/pack_info/pack_pos"].value
if body_size:
yield RawBytes(self, "compressed_data", body_size, "Compressed data")
# Here we could check if copy method was used to "compress" it,
# but this never happens, so just output "compressed file info"
yield get_header_field(self, "compressed_file_info", pack_size,
"Compressed file information")
size = (self._size // 8) - pack_size - body_size
if size > 0:
yield RawBytes(self, "unknown_data", size)
elif "header" in self["/next_hdr"]:
yield RawBytes(self, "compressed_data", self._size // 8, "Compressed data")
class StartHeader(FieldSet):
static_size = 160
def createFields(self):
yield textHandler(UInt64(self, "next_hdr_offset",
"Next header offset"), hexadecimal)
yield UInt64(self, "next_hdr_size", "Next header size")
yield textHandler(UInt32(self, "next_hdr_crc",
"Next header CRC"), hexadecimal)
class SignatureHeader(FieldSet):
static_size = 96 + StartHeader.static_size
def createFields(self):
yield Bytes(self, "signature", 6, "Signature Header")
yield UInt8(self, "major_ver", "Archive major version")
yield UInt8(self, "minor_ver", "Archive minor version")
yield textHandler(UInt32(self, "start_hdr_crc",
"Start header CRC"), hexadecimal)
yield StartHeader(self, "start_hdr", "Start header")
class SevenZipParser(Parser):
MAGIC = b"7z\xbc\xaf\x27\x1c"
PARSER_TAGS = {
"id": "7zip",
"category": "archive",
"file_ext": ("7z",),
"mime": ("application/x-7z-compressed",),
"min_size": 32 * 8,
"magic": ((MAGIC, 0),),
"description": "Compressed archive in 7z format"
}
endian = LITTLE_ENDIAN
def createFields(self):
yield SignatureHeader(self, "signature", "Signature Header")
yield Body(self, "body_data")
yield NextHeader(self, "next_hdr")
def validate(self):
if self.stream.readBytes(0, len(self.MAGIC)) != self.MAGIC:
return "Invalid signature"
return True
def createContentSize(self):
size = self["/signature/start_hdr/next_hdr_offset"].value * 8
size += self["/signature/start_hdr/next_hdr_size"].value * 8
size += SignatureHeader.static_size
return size
Reference in a new issue Copy permalink