From ab9cc7b196efd62f6976943d15e8edebca76bf56 Mon Sep 17 00:00:00 2001 From: Dobin Rutishauser Date: Fri, 29 Sep 2023 10:22:12 +0200 Subject: [PATCH] refactor: phase 1 of pe section rework --- model/file_model.py | 6 ++- model/model_code.py | 15 +++--- plugins/dotnet/augment_dotnet.py | 17 +++---- plugins/pe/analyzer_pe.py | 41 +++------------ plugins/pe/augment_pe.py | 10 ++-- plugins/pe/file_pe.py | 86 +++++++++++++------------------- tests/test_dotnet.py | 8 +-- tests/test_modelcode.py | 24 +++++++++ 8 files changed, 93 insertions(+), 114 deletions(-) create mode 100644 tests/test_modelcode.py diff --git a/model/file_model.py b/model/file_model.py index 64729ef..b1dc2f7 100644 --- a/model/file_model.py +++ b/model/file_model.py @@ -13,8 +13,10 @@ class BaseFile(): self.fileData: Data = Data(b'') # The content of the file self.data: Data = Data(b'') # The data we work on - self.peSectionsBag: SectionsBag = SectionsBag() # cover complete file - self.regionsBag: SectionsBag = SectionsBag() # more details + self.peSectionsBag: SectionsBag = SectionsBag() # cover complete file, saved, overlap + self.dotnetSectionsBag: SectionsBag = SectionsBag() # for DotNet "sections" + self.scanSectionsBag: SectionsBag = SectionsBag() # the sections to scan, not-saved, no-overlap + self.regionsBag: SectionsBag = SectionsBag() # more details saved, no-overlap def parseFile(self) -> bool: diff --git a/model/model_code.py b/model/model_code.py index a27340d..8f7749d 100644 --- a/model/model_code.py +++ b/model/model_code.py @@ -66,8 +66,14 @@ class SectionsBag: if address >= section.virtaddr and address <= section.virtaddr + section.size: return section return None - + def getSectionNameByPhysAddr(self, address: int) -> Section: + for section in self.sections: + if address >= section.physaddr and address <= section.physaddr + section.size: + return section.name + return "" + + def containsSectionName(self, address: int, name: str) -> bool: """Returns true if one of the section names of address is name""" for section in self.sections: @@ -75,13 +81,6 @@ class SectionsBag: if section.name == name: return True return False - - - def getSectionNameByPhysAddr(self, address: int) -> Section: - for section in self.sections: - if address >= section.physaddr and address <= section.physaddr + section.size: - return section.name - return "" def getSectionsForPhysRange(self, start: int, end: int) -> List[Section]: diff --git a/plugins/dotnet/augment_dotnet.py b/plugins/dotnet/augment_dotnet.py index 7205107..c87aea2 100644 --- a/plugins/dotnet/augment_dotnet.py +++ b/plugins/dotnet/augment_dotnet.py @@ -36,20 +36,19 @@ def augmentFileDotnet(filePe: FilePe, matches: List[Match]) -> str: #if dotnetSectionsBag is not None: # set info: .NET sections/streams name next if found - sections = filePe.peSectionsBag.getSectionsForPhysRange(match.start(), match.end()) + info = filePe.peSectionsBag.getSectionByPhysAddr(match.start()).name + info += " " + filePe.dotnetSectionsBag.getSectionByPhysAddr(match.start()).name regions = filePe.regionsBag.getSectionsForPhysRange(match.start(), match.end()) - info = ' '.join(s.name for s in sections) info += ' '.join(s.name for s in regions) if filePe.peSectionsBag.containsSectionName(match.fileOffset, ".text"): - # only disassemble if the match is reasonably small. same for function names - if match.size < MAX_DISASM_SIZE: - matchAsmInstructions, matchDisasmLines, methodNames = disassembleDotNet(match.start(), match.size, dncilParser) - detail += " " + " ".join(methodNames) - # .text has most of DotNet, check if its methods - if filePe.peSectionsBag.containsSectionName(match.fileOffset, "methods"): + if filePe.dotnetSectionsBag.containsSectionName(match.fileOffset, "methods"): match.sectionType = SectionType.CODE + if match.size < MAX_DISASM_SIZE: + # only disassemble if the match is reasonably small. same for function names + matchAsmInstructions, matchDisasmLines, methodNames = disassembleDotNet(match.start(), match.size, dncilParser) + detail += " " + " ".join(methodNames) else: match.sectionType = SectionType.DATA results: List[DotnetDataEntry] = dotNetData.findBy(match.start(), match.end()) @@ -72,7 +71,7 @@ def augmentFileDotnet(filePe: FilePe, matches: List[Match]) -> str: match.sectionType = SectionType.DATA # take dotnet section - relevantSection = filePe.peSectionsBag.getSectionByPhysAddr(match.fileOffset) + relevantSection = filePe.dotnetSectionsBag.getSectionByPhysAddr(match.fileOffset) #if relevantSection is None: # # or pe section if not in dotnet section # relevantSection = filePe.peSectionsBag.getSectionByPhysAddr(match.fileOffset) diff --git a/plugins/pe/analyzer_pe.py b/plugins/pe/analyzer_pe.py index 6783869..96b7b2a 100644 --- a/plugins/pe/analyzer_pe.py +++ b/plugins/pe/analyzer_pe.py @@ -15,13 +15,12 @@ from scanning import scanIsHash def analyzeFilePe(filePe: FilePe, scanner: Scanner, reducer: Reducer, analyzerOptions={}) -> Tuple[Match, ScanInfo]: """Scans a PE file given with filePe with Scanner scanner. Returns all matches and ScanInfo""" - isolate = analyzerOptions.get("isolate", False) scanSpeed = analyzerOptions.get("scanSpeed", ScanSpeed.Normal) scanInfo = ScanInfo(scanner.scanner_name, scanSpeed) # prepare the reducer with the file timeStart = time.time() - matches, scanPipe = scanForMatchesInPe(filePe, scanner, reducer, isolate) + matches, scanPipe = scanForMatchesInPe(filePe, scanner, reducer) scanInfo.scanDuration = round(time.time() - timeStart) scanInfo.scannerPipe = scanPipe scanInfo.chunksTested = reducer.chunks_tested @@ -30,7 +29,7 @@ def analyzeFilePe(filePe: FilePe, scanner: Scanner, reducer: Reducer, analyzerOp return matches, scanInfo -def scanForMatchesInPe(filePe: FilePe, scanner: Scanner, reducer: Reducer, isolate=False) -> Tuple[List[Match], str]: +def scanForMatchesInPe(filePe: FilePe, scanner: Scanner, reducer: Reducer) -> Tuple[List[Match], str]: """Scans a PE file given with filePe with Scanner scanner. Returns all matches""" scanStages = [] matches: List[Match] = [] @@ -38,14 +37,8 @@ def scanForMatchesInPe(filePe: FilePe, scanner: Scanner, reducer: Reducer, isola # identify which sections get detected # default is to not-isolate detected_sections = [] - if isolate: - logging.info("Section Detection: Isolating sections (zero all others)") - scanStages.append('ident:zero-nontarget-sections') - detected_sections = findDetectedSectionsIsolate(filePe, scanner) - else: - logging.info("Section Detection: Zero section (leave all others intact)") - scanStages.append('ident:zero-target-section') - detected_sections = findDetectedSections(filePe, scanner) + logging.info("Section Detection: Zero section (leave all others intact)") + detected_sections = findDetectedSections(filePe, scanner) logging.info(f"{len(detected_sections)} section(s) trigger the antivirus independantly") for section in detected_sections: logging.info(f" section: {section.name}") @@ -97,13 +90,13 @@ def findDetectedSections(filePe: FilePe, scanner) -> List[Section]: """hide each section of filePe, return the ones which wont be detected anymore (have a dominant influence)""" detected_sections: List[Section] = [] - for idx, section in enumerate(filePe.peSectionsBag.sections): + for idx, section in enumerate(filePe.scanSectionsBag.sections): filePeCopy = deepcopy(filePe) filePeCopy.hideSection(section) detected = scanner.scannerDetectsBytes(filePeCopy.DataAsBytes(), filePeCopy.filename) if not detected: # always store scan result - filePe.peSectionsBag.sections[idx].detected = True + filePe.scanSectionsBag.sections[idx].detected = True # only return if we should scan it if section.scan: @@ -112,25 +105,3 @@ def findDetectedSections(filePe: FilePe, scanner) -> List[Section]: logging.info(f"Hide: {section.name} -> Detected: {detected} (to scan: {section.scan})") return detected_sections - - -def findDetectedSectionsIsolate(filePe: FilePe, scanner) -> List[Section]: - """for each section, hide everything except it (isolate), and return which one gets detected (have a dominant influence)""" - detected_sections = [] - - for idx, section in enumerate(filePe.peSectionsBag.sections): - filePeCopy = deepcopy(filePe) - filePeCopy.hideAllSectionsExcept(section.name) - detected = scanner.scannerDetectsBytes(filePeCopy.DataAsBytes(), filePeCopy.filename) - - if not detected: - # always store scan result - filePe.peSectionsBag.sections[idx].detected = True - - # only return if we should scan it - if section.scan: - detected_sections += [section] - - logging.info(f"Hide all except: {section.name} -> Detected: {detected} (to scan: {section.scan})") - - return detected_sections diff --git a/plugins/pe/augment_pe.py b/plugins/pe/augment_pe.py index f72b127..af85ba0 100644 --- a/plugins/pe/augment_pe.py +++ b/plugins/pe/augment_pe.py @@ -116,11 +116,11 @@ def augmentFilePe(filePe: FilePe, matches: List[Match]) -> str: match.setAsmInstructions(matchAsmInstructions) # file structure - s = '' - for matchSection in filePe.peSectionsBag.sections: - s += "{0:<16}: File Offset: {1:<7} Virtual Addr: {2:<6} size {3:<6} scan:{4}\n".format( - matchSection.name, matchSection.physaddr, matchSection.virtaddr, matchSection.size, matchSection.scan) - return s + #s = '' + #for matchSection in filePe.peSectionsBag.sections: + # s += "{0:<16}: File Offset: {1:<7} Virtual Addr: {2:<6} size {3:<6} scan:{4}\n".format( + # matchSection.name, matchSection.physaddr, matchSection.virtaddr, matchSection.size, matchSection.scan) + #return s conv = Ansi2HTMLConverter() diff --git a/plugins/pe/file_pe.py b/plugins/pe/file_pe.py index 62a8dbc..c1ec925 100644 --- a/plugins/pe/file_pe.py +++ b/plugins/pe/file_pe.py @@ -38,13 +38,27 @@ class FilePe(BaseFile): # handle dotnet if self.isDotNet: - # self.peSectionsBag + # self.dotnetSectionsBag self.parseDotNetSections() # added DotNet specific sections in .text, so disable scanning of .text self.peSectionsBag.getSectionByName(".text").scan = False # self.regionsBag - self.parseDotNetRegions() + #self.parseDotNetRegions() + + # peSectionsBag is built. get the ones we want to scan + self.scanSectionsBag = self.getScanSections() + + + def getScanSections(self): + bag = SectionsBag() + for section in self.peSectionsBag.sections: + if section.scan: + bag.addSection(section) + for section in self.dotnetSectionsBag.sections: + if section.scan: + bag.addSection(section) + return bag def parsePeSections(self, pepe, fileLength): @@ -93,35 +107,6 @@ class FilePe(BaseFile): )) - def parseDotNetRegions(self): - logging.info("FilePe: Parse DotNet Regions") - dotnet_file = DotNetPE(self.filepath) - textSection: Section = self.peSectionsBag.getSectionByName('.text') - addrOffset = textSection.virtaddr - textSection.physaddr - - # metadata header - metadata_header_addr = dotnet_file.dotnet_metadata_header.address - addrOffset - metadata_header_vaddr = dotnet_file.dotnet_metadata_header.address - metadata_header_size = dotnet_file.dotnet_metadata_header.size - - # Metadata header - s = Section('Metadata Header', - metadata_header_addr, - metadata_header_size, - metadata_header_vaddr, - False) - self.regionsBag.addSection(s) - - # All stream headers - for streamHeader in dotnet_file.dotnet_stream_headers: - s = Section(streamHeader.string_representation, - streamHeader.address - addrOffset, - streamHeader.size, - streamHeader.address, - False) - self.regionsBag.addSection(s) - - def rvaToPhysOffset(self, rva: int) -> int: section = self.peSectionsBag.getSectionByVirtAddr(rva) if section is None: @@ -180,28 +165,27 @@ class FilePe(BaseFile): cli_header_size, cli_header_vaddr, False) - self.peSectionsBag.addSection(s) - - # more header: - # * start at metadata header - # * stop at metadata stream start - # (usually?) - more_header_addr = dotnet_file.dotnet_metadata_header.address - addrOffset - more_header_vaddr = dotnet_file.dotnet_metadata_header.address - more_header_size = dotnet_file.dotnet_metadata_header.size - for streamHeader in dotnet_file.dotnet_stream_headers: - more_header_size += streamHeader.size - s = Section('MoreHeader', - more_header_addr, - more_header_size, - more_header_vaddr, - scan = False) - self.peSectionsBag.addSection(s) + self.dotnetSectionsBag.addSection(s) # metadata header metadata_header_addr = dotnet_file.dotnet_metadata_header.address - addrOffset metadata_header_vaddr = dotnet_file.dotnet_metadata_header.address metadata_header_size = dotnet_file.dotnet_metadata_header.size + s = Section('Metadata Header', + metadata_header_addr, + metadata_header_size, + metadata_header_vaddr, + False) + self.dotnetSectionsBag.addSection(s) + + # All stream headers + for streamHeader in dotnet_file.dotnet_stream_headers: + s = Section(streamHeader.string_representation, + streamHeader.address - addrOffset, + streamHeader.size, + streamHeader.address, + False) + self.dotnetSectionsBag.addSection(s) # methods methods_addr = cli_header_addr + cli_header_size @@ -211,7 +195,7 @@ class FilePe(BaseFile): methods_addr, methods_size, methods_vaddr) - self.peSectionsBag.addSection(s) + self.dotnetSectionsBag.addSection(s) # metadata directory #metadata_directory_addr = dotnet_file.clr_header.MetaDataDirectoryAddress.value @@ -234,7 +218,7 @@ class FilePe(BaseFile): signature_size, 0, False) - self.peSectionsBag.addSection(s) + self.dotnetSectionsBag.addSection(s) # All streams stream: FileLocation @@ -243,7 +227,7 @@ class FilePe(BaseFile): stream.address - addrOffset, stream.size, stream.address) - self.peSectionsBag.addSection(s) + self.dotnetSectionsBag.addSection(s) # Name correspond to their index diff --git a/tests/test_dotnet.py b/tests/test_dotnet.py index f7ef10c..1da6ff8 100644 --- a/tests/test_dotnet.py +++ b/tests/test_dotnet.py @@ -116,11 +116,11 @@ class DotnetDisasmTest(unittest.TestCase): filePe.loadFromFile("tests/data/HelloWorld.dll") self.assertTrue(filePe.isDotNet) - section = filePe.peSectionsBag.getSectionByName('DotNet Header') + section = filePe.dotnetSectionsBag.getSectionByName('DotNet Header') self.assertEqual(section.physaddr, 512) self.assertEqual(section.size, 72) - section = filePe.peSectionsBag.getSectionByName('methods') + section = filePe.dotnetSectionsBag.getSectionByName('methods') self.assertEqual(section.physaddr, 584) self.assertEqual(section.size, 28) @@ -130,7 +130,7 @@ class DotnetDisasmTest(unittest.TestCase): filePe.loadFromFile("tests/data/HelloWorld.dll") self.assertTrue(filePe.isDotNet) - section = filePe.regionsBag.getSectionByName('#~ Stream Header') + section = filePe.dotnetSectionsBag.getSectionByName('#~ Stream Header') self.assertEqual(section.physaddr, 644) self.assertEqual(section.virtaddr, 0x2084) self.assertEqual(section.size, 12) @@ -191,6 +191,6 @@ class DotnetDisasmTest(unittest.TestCase): filePe.loadFromFile("tests/data/HelloWorld-signed.dll") self.assertTrue(filePe.isDotNet) - section = filePe.peSectionsBag.getSectionByName("Signature") + section = filePe.dotnetSectionsBag.getSectionByName("Signature") self.assertEqual(section.physaddr, 2088) self.assertEqual(section.size, 128) diff --git a/tests/test_modelcode.py b/tests/test_modelcode.py new file mode 100644 index 0000000..fdb48c0 --- /dev/null +++ b/tests/test_modelcode.py @@ -0,0 +1,24 @@ +import unittest + +from plugins.pe.file_pe import FilePe +from plugins.pe.analyzer_pe import analyzeFilePe +from plugins.pe.augment_pe import augmentFilePe, disassemblePe +from plugins.pe.outflank_pe import outflankPe +from tests.helpers import TestDetection +from tests.scanners import * +from model.model_data import Match +from model.model_verification import MatchConclusion, VerifyStatus +from myutils import hexdmp, hexstr, removeAnsi +from reducer import Reducer + + +class TestModelCode(unittest.TestCase): + def test_modelcode(self): + filePe = FilePe() + + filePe.peSectionsBag.getSectionByName() + filePe.peSectionsBag.getSectionByPhysAddr() + filePe.peSectionsBag.getSectionByVirtAddr() + filePe.peSectionsBag.containsSectionName() + filePe.peSectionsBag.getSectionNameByPhysAddr() + filePe.peSectionsBag.getSectionsForPhysRange()