def run(self):
"""Run analysis.
@return: analysis results dict or None.
"""
if not os.path.exists(self.file_path):
return {}
try:
self.pe = pefile.PE(self.file_path)
except pefile.PEFormatError:
return {}
results = {}
results["peid_signatures"] = self._get_peid_signatures()
results["pe_imports"] = self._get_imported_symbols()
results["pe_exports"] = self._get_exported_symbols()
results["pe_sections"] = self._get_sections()
results["pe_resources"] = self._get_resources()
results["pe_versioninfo"] = self._get_versioninfo()
results["pe_imphash"] = self._get_imphash()
results["pe_timestamp"] = self._get_timestamp()
results["pdb_path"] = self._get_pdb_path()
results["signature"] = self._get_signature()
results["imported_dll_count"] = len([x for x in results["pe_imports"] if x.get("dll")])
return results
python类PEFormatError()的实例源码
def initialize(self, sample):
if(self.already_initialized):
return self.library
self.already_initialized = True
try:
self.library = pefile.PE(data=sample.getBinary(), fast_load=True)
# see if this initializations can be done on plugins.
self.library.parse_data_directories(directories=[
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_IMPORT'],
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_EXPORT'],
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_TLS'],
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_SECURITY'],
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_RESOURCE']])
except pefile.PEFormatError:
# print("parse fail")
self.library = None
# print(traceback.format_exc())
logging.error("Error parsing pefileModule with sample:%s",
sample.getID(), exc_info=True)
def __read_props(self):
def _loword(dword):
return dword & 0x0000ffff
def _hiword(dword):
return dword >> 16
self.__props = {}
try:
pe = pefile.PE(self.__path)
except pefile.PEFormatError as e:
raise ValueError(e.value)
else:
ms = pe.VS_FIXEDFILEINFO.ProductVersionMS
ls = pe.VS_FIXEDFILEINFO.ProductVersionLS
self.__props['fixed_version'] = '.'.join(map(str, (_hiword(ms), _loword(ms), _hiword(ls), _loword(ls))))
for file_info in pe.FileInfo:
if file_info.Key == b'StringFileInfo':
for st in file_info.StringTable:
for entry in st.entries.items():
self.__props[entry[0].decode('latin_1')] = entry[1].decode('latin_1')
# noinspection PyAbstractClass
def test_nt_headers_exception(self):
"""pefile should fail parsing invalid data (missing NT headers)"""
# Take a known good file.
control_file = os.path.join(REGRESSION_TESTS_DIR, 'MSVBVM60.DLL')
pe = pefile.PE(control_file, fast_load=True)
# Truncate it at the PE header and add invalid data.
pe_header_offest = pe.DOS_HEADER.e_lfanew
corrupted_data = pe.__data__[:pe_header_offest] + b'\0' * (1024 * 10)
self.assertRaises(pefile.PEFormatError, pefile.PE, data=corrupted_data)
def test_dos_header_exception_large_data(self):
"""pefile should fail parsing 10KiB of invalid data
(missing DOS header).
"""
# Generate 10KiB of zeroes
data = b'\0' * (1024 * 10)
# Attempt to parse data and verify PE header, a PEFormatError exception
# is thrown.
self.assertRaises(pefile.PEFormatError, pefile.PE, data=data)
def test_dos_header_exception_small_data(self):
"""pefile should fail parsing 64 bytes of invalid data
(missing DOS header).
"""
# Generate 64 bytes of zeroes
data = b'\0' * (64)
# Attempt to parse data and verify PE header a PEFormatError exception
# is thrown.
self.assertRaises(pefile.PEFormatError, pefile.PE, data=data)
def test_empty_file_exception(self):
"""pefile should fail parsing empty files."""
# Take a known good file
control_file = os.path.join(REGRESSION_TESTS_DIR, 'empty_file')
self.assertRaises(pefile.PEFormatError, pefile.PE, control_file)
def ScanFile(filename, signatures, minimumEntropy):
global oLogger
if not FileContentsStartsWithMZ(filename):
return
try:
pe = GetPEObject(filename)
except pefile.PEFormatError:
oLogger.PrintAndLog(('%s', '%s'), (filename, 'PEFormatError'))
return
except TypeError:
oLogger.PrintAndLog(('%s', '%s'), (filename, 'TypeError'))
return
try:
raw = pe.write()
except MemoryError:
oLogger.PrintAndLog(('%s', '%s'), (filename, 'MemoryError'))
return
entropy = pe.sections[0].entropy_H(raw)
if entropy >= minimumEntropy:
countFlagsExecute = 0
countFlagsExecuteAndWrite = 0
for section in pe.sections:
if section.IMAGE_SCN_MEM_EXECUTE:
countFlagsExecute += 1
if section.IMAGE_SCN_MEM_EXECUTE and section.IMAGE_SCN_MEM_WRITE:
countFlagsExecuteAndWrite += 1
calculatedCRC = pe.generate_checksum()
crcDifferent = pe.OPTIONAL_HEADER.CheckSum != 0 and pe.OPTIONAL_HEADER.CheckSum != calculatedCRC
info = GetVersionInfo(pe)
oLogger.PrintAndLog(('%s', '%f', '%d', '%d', '%d', '%d', '%08X', '%08X', '%d', '%s', '%s', '%s', '%s'), (filename, entropy, len(pe.sections), countFlagsExecute, countFlagsExecuteAndWrite, pe.OPTIONAL_HEADER.DATA_DIRECTORY[pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_SECURITY']].Size, pe.OPTIONAL_HEADER.CheckSum, calculatedCRC, crcDifferent, time.asctime(time.gmtime(pe.FILE_HEADER.TimeDateStamp)), repr(RVOES(info, 'CompanyName')), repr(RVOES(info, 'ProductName')), hashlib.md5(raw).hexdigest()))
def __check_session(self):
if not __sessions__.is_set():
self.log('error', "No open session")
return False
if not self.pe:
try:
self.pe = pefile.PE(__sessions__.current.file.path)
except pefile.PEFormatError as e:
self.log('error', "Unable to parse PE file: {0}".format(e))
return False
return True
def getArquitecture(self):
try:
if(self.pe.OPTIONAL_HEADER.Magic == int("0x020B", 16)):
return ("PE+")
elif(self.pe.OPTIONAL_HEADER.Magic == int("0x010B", 16)):
return ("PE")
elif(self.pe.OPTIONAL_HEADER.Magic == int("0x0107", 16)):
return ("IMG_ROM")
else:
return "UNKNOWN"
except pefile.PEFormatError:
return "FORMAT"
return None
def __check_session(self):
if not __sessions__.is_set():
self.log('error', "No open session")
return False
if not self.pe:
try:
self.pe = pefile.PE(__sessions__.current.file.path)
except pefile.PEFormatError as e:
self.log('error', "Unable to parse PE file: {0}".format(e))
return False
return True
def check_pe_header(filepath):
try:
pe = pefile.PE(filepath)
if (pe.DOS_HEADER.e_magic == int(0x5a4d) and pe.NT_HEADERS.Signature == int(0x4550)):
return True
except (pefile.PEFormatError):
return False
except(Exception) as e:
print("LOG - Something weird %s - %s" % (str(e), filepath))
return False
return False
def process(self):
pelib = self._getLibrary(PEFileModule().getName())
if(pelib is None):
return ""
ret = []
if hasattr(pelib, 'DIRECTORY_ENTRY_RESOURCE'):
i = 0
for resource_type in pelib.DIRECTORY_ENTRY_RESOURCE.entries:
if resource_type.name is not None:
name = "%s" % resource_type.name
else:
name = "%s" % pefile.RESOURCE_TYPE.get(
resource_type.struct.Id)
if name is None:
name = "%d" % resource_type.struct.Id
if hasattr(resource_type, 'directory'):
for resource_id in resource_type.directory.entries:
if hasattr(resource_id, 'directory'):
for resource_lang in resource_id.directory.entries:
try:
data = pelib.get_data(
resource_lang.data.struct.OffsetToData, resource_lang.data.struct.Size)
# fd=open(name,'wb')
# fd.write(data)
# (data)
except pefile.PEFormatError:
return "corrupt"
filetype = MIME_TYPE(data, False)
lang = pefile.LANG.get(
resource_lang.data.lang, 'unknown')
sublang = pefile.get_sublang_name_for_lang(
resource_lang.data.lang, resource_lang.data.sublang)
entry = {}
entry["name"] = self._normalize(name)
entry["rva"] = self._normalize(
hex(resource_lang.data.struct.OffsetToData))
entry["size"] = self._normalize(
hex(resource_lang.data.struct.Size))
entry["type"] = self._normalize(filetype)
entry["lang"] = self._normalize(lang)
entry["sublang"] = self._normalize(sublang)
entry["sha1"] = SHA1(data)
ret.append(entry)
return ret
def main():
parser = argparse.ArgumentParser(description='Process pehash in different ways')
parser.add_argument('--totalhash', dest='totalhash', action='store_true',
default=False, help="Generate totalhash pehash")
parser.add_argument('--anymaster', dest='anymaster', action='store_true',
default=False, help="Generate anymaster pehash")
parser.add_argument('--anymaster_v1', dest='anymaster_v1', action='store_true',
default=False, help="Generate anymaster v1.0.1 pehash")
parser.add_argument('--endgame', dest='endgame', action='store_true',
default=False, help="Generate endgame pehash")
parser.add_argument('--crits', dest='crits', action='store_true',
default=False, help="Generate crits pehash")
parser.add_argument('--pehashng', dest='pehashng', action='store_true',
default=False, help="Generate pehashng (https://github.com/AnyMaster/pehashng)")
parser.add_argument('-v', dest='verbose', action='count', default=0,
help="Raise pehash exceptions instead of ignoring.")
parser.add_argument('binaries', metavar='binaries', type=str, nargs='+',
help='list of pe files to process')
args = parser.parse_args()
for binary in args.binaries:
try:
pe = pefile.PE(binary)
do_all = not (args.totalhash or args.anymaster or args.anymaster_v1 or
args.endgame or args.crits or args.pehashng)
raise_on_error = True if args.verbose > 0 else False
if args.totalhash or do_all:
print("{}\tTotalhash\t{}".format(binary, totalhash_hex(pe=pe, raise_on_error=raise_on_error)))
if args.anymaster or do_all:
print("{}\tAnyMaster\t{}".format(binary, anymaster_hex(pe=pe, raise_on_error=raise_on_error)))
if args.anymaster_v1 or do_all:
print("{}\tAnyMaster_v1.0.1\t{}".format(binary, anymaster_v1_0_1_hex(pe=pe, raise_on_error=raise_on_error)))
if args.endgame or do_all:
print("{}\tEndGame\t{}".format(binary, endgame_hex(pe=pe, raise_on_error=raise_on_error)))
if args.crits or do_all:
print("{}\tCrits\t{}".format(binary, crits_hex(pe=pe, raise_on_error=raise_on_error)))
if args.pehashng or do_all:
print("{}\tpeHashNG\t{}".format(binary, pehashng_hex(pe=pe, raise_on_error=raise_on_error)))
except pefile.PEFormatError:
print("ERROR: {} is not a PE file".format(binary))
def main():
"""
arg1 - path to files
arg2 - path to res dir
"""
ben_files = []
corrupted_files = []
for file_name in os.listdir(sys.argv[1]):
path_to_file = join(sys.argv[1], file_name)
if not os.path.isfile(path_to_file): continue
print "Try to ent-split file", file_name
try:
ben_files.append([file_name, build_tree(
get_execution_block(path_to_file))])
except pefile.PEFormatError:
print "Can't process file"
corrupted_files.append(file_name)
for ben_file in ben_files:
ben_file.append(quad_fourier(ben_file[1]))
ben_sorted_groups = []
for ben_file in ben_files:
was_found = False
if(ben_file[2][0] == 0.0): continue
for group in ben_sorted_groups:
if abs(group[0][0] - ben_file[2][0]) / group[0][0] < limit and \
abs(group[0][1] - ben_file[2][1]) / group[0][1] < limit:
was_found = True
group[1].append(ben_file[0])
break
if not was_found:
ben_sorted_groups.append([ben_file[2], [ben_file[0]]])
unsorted_groups = filter(lambda x: len(x[1]) == 1, ben_sorted_groups)
ben_sorted_groups = filter(lambda x: len(x[1]) > 1, ben_sorted_groups)
res_dir = sys.argv[2]
for group in ben_sorted_groups:
dir_name = join(res_dir, "[{0}]V={1},E={2}".format(len(group[1]), "%.2f"%group[0][0], "%.2f"%group[0][1]))
os.mkdir(join(dir_name))
for file_name in group[1]:
shutil.copyfile(join(sys.argv[1], file_name), join(dir_name, os.path.basename(file_name)))
def get_attr_pe(r, sha256):
path = r.hget(sha256, 'path')
try:
pe = pefile.PE(path)
except (pefile.PEFormatError):
print("{} not a PE file".format(path))
return False
r.hset(sha256, 'is_pefile', True)
if hasattr(pe, 'FILE_HEADER'):
r.hset(sha256, 'timestamp', pe.FILE_HEADER.TimeDateStamp)
r.hset(sha256, 'timestamp_iso', datetime.datetime.fromtimestamp(pe.FILE_HEADER.TimeDateStamp).isoformat())
r.zincrby('timestamps', pe.FILE_HEADER.TimeDateStamp)
r.sadd('timestamp:{}'.format(pe.FILE_HEADER.TimeDateStamp), sha256)
imphash = pe.get_imphash()
r.hset(sha256, 'imphash', imphash)
r.zincrby('imphashs', imphash)
r.sadd('imphash:{}'.format(imphash), sha256)
if hasattr(pe, 'OPTIONAL_HEADER'):
r.hset(sha256, 'entrypoint', pe.OPTIONAL_HEADER.AddressOfEntryPoint)
r.zincrby('entrypoints', pe.OPTIONAL_HEADER.AddressOfEntryPoint)
r.sadd('entrypoint:{}'.format(pe.OPTIONAL_HEADER.AddressOfEntryPoint), sha256)
if hasattr(pe, 'FILE_HEADER'):
r.hset(sha256, 'secnumber', pe.FILE_HEADER.NumberOfSections)
r.zincrby('secnumbers', pe.FILE_HEADER.NumberOfSections)
r.sadd('secnumber:{}'.format(pe.FILE_HEADER.NumberOfSections), sha256)
if hasattr(pe, 'VS_VERSIONINFO'):
for entry in pe.FileInfo:
if hasattr(entry, 'StringTable'):
for st_entry in entry.StringTable:
ofn = st_entry.entries.get(b'OriginalFilename')
if ofn:
if isinstance(ofn, bytes):
o = ofn.decode()
else:
o = ofn
r.hset(sha256, 'originalfilename', o)
r.zincrby('originalfilenames', o)
r.sadd(u'originalfilename:{}'.format(o), sha256)
# Section info: names, sizes, entropy vals
for section in pe.sections:
name = section.Name.decode('utf-8', 'ignore').replace('\x00', '')
r.sadd('{}:secnames'.format(sha256), name)
r.hset('{}:{}'.format(sha256, name), 'size', section.SizeOfRawData)
r.hset('{}:{}'.format(sha256, name), 'entropy', H(section.get_data()))
# adding section info to PE data
r.hset(sha256, 'nb_tls', check_tls(pe))
r.hset(sha256, 'ep_section', check_ep_section(pe))
return True
# Returns Entropy value for given data chunk
