diff --git a/model/model.py b/model/model.py index 30f4347..5a1a3f4 100644 --- a/model/model.py +++ b/model/model.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from dataclasses import dataclass from typing import List, Set, Dict, Tuple, Optional from enum import Enum @@ -137,15 +139,30 @@ class SectionsBag: print(f"Section {section.name}\t addr: {hex(section.addr)} size: {section.size} ") +gpRegisters = [ 'rax','rbx','rcx','rdx','rsi','rdi','r8','r9','r10','r11','r12','r13','r14','r15' ] class AsmInstruction(): - def __init__(self, fileOffset, rva, esil, type, disasm, size, bytes): + def __init__(self, fileOffset, rva, esil, type, disasm, size, rawBytes): self.offset = fileOffset # offset in file self.rva = rva self.esil = esil self.type = type self.disasm = disasm self.size = size - self.bytes = bytes + self.rawBytes = rawBytes + + # ESIL + self.esilComponents = esil.split(",") if esil else [] + + # Registers touched + self.esilTouchedRegisters = [] + for a in self.esilComponents: + if a in gpRegisters: + self.esilTouchedRegisters.append(a) + + + def registersTouch(self, asmInstruction: AsmInstruction): + return any(element in self.esilTouchedRegisters for element in asmInstruction.esilTouchedRegisters) + def __str__(self): s = "Offset: {} RVA: {} type: {} disasm: {} size: {} esil: {} bytes: {}".format( @@ -155,7 +172,7 @@ class AsmInstruction(): self.disasm, self.size, self.esil, - self.bytes + self.rawBytes ) return s @@ -297,12 +314,25 @@ class Outcome(): class OutflankPatch(): - def __init__(self, matchIdx: int, offset: int, replaceBytes: bytes, info: str, considerations: str): + def __init__(self, + matchIdx: int, + offset: int, + replaceBytes: bytes, + asmOne: AsmInstruction, + asmTwo: AsmInstruction, + info: str, + considerations: str + ): self.matchIdx = matchIdx self.offset = offset self.replaceBytes = replaceBytes + + self.asmOne = asmOne + self.asmTwo = asmTwo + self.info = info self.considerations = considerations + def __str__(self): s = '' diff --git a/plugins/analyzer_pe.py b/plugins/analyzer_pe.py index c356d84..751f2dd 100644 --- a/plugins/analyzer_pe.py +++ b/plugins/analyzer_pe.py @@ -89,7 +89,7 @@ def disassemble(r2, filePe: FilePe, fileOffset: int, sizeDisasm: int, moreUiLine type = a.get('type', '') disasm = a.get('disasm', '') size = a.get('size', 0) - bytes = a.get('bytes', b'') + rawBytes = bytes(a.get('bytes', ''), 'utf-8') asmInstruction = AsmInstruction( asmFileOffset, @@ -98,7 +98,7 @@ def disassemble(r2, filePe: FilePe, fileOffset: int, sizeDisasm: int, moreUiLine type, disasm, size, - bytes) + rawBytes) matchAsmInstructions.append(asmInstruction) virtAddrDisasm -= moreUiLines @@ -128,7 +128,7 @@ def disassemble(r2, filePe: FilePe, fileOffset: int, sizeDisasm: int, moreUiLine textHtml, ) matchDisasmLines.append(disasmLine) - + return matchAsmInstructions, matchDisasmLines diff --git a/plugins/outflank_pe.py b/plugins/outflank_pe.py index cb11a63..259644f 100644 --- a/plugins/outflank_pe.py +++ b/plugins/outflank_pe.py @@ -1,12 +1,12 @@ from typing import List, Set, Dict, Tuple, Optional -import re + import logging -from model.model import Outcome, OutflankPatch, Match, MatchConclusion, Data +from model.model import Outcome, OutflankPatch, Match, MatchConclusion, Data, AsmInstruction from model.testverify import VerifyStatus from model.extensions import Scanner from plugins.file_pe import FilePe - +from utils import removeAnsi class PossiblePatch(): def __init__(self, offset, matchId): @@ -19,90 +19,65 @@ def outflankPe( ) -> List[OutflankPatch]: results: List[OutflankPatch] = [] - #for line in matches[0].disasmLines: - # print(escape_ansi(str(line))) - - nopLines: List[PossiblePatch] = [] - int3Lines: List[PossiblePatch] = [] for idx, match in enumerate(matches): - if matchConclusion.verifyStatus[idx] != VerifyStatus.GOOD: - continue + #if matchConclusion.verifyStatus[idx] != VerifyStatus.GOOD: + # continue - disasmLines = match.getDisasmLines() - for line in disasmLines: - if not line.isPart: - continue + asm: AsmInstruction + n = 0 + while n < len(match.asmInstructions) - 1: + asm = match.asmInstructions[n] + nextAsm = match.asmInstructions[n+1] + print(asm) - s = escape_ansi(str(line)) - if '90 nop' in s: - nopLines.append(PossiblePatch(line.offset, idx)) + if (asm.type == "nop" and nextAsm.type == "nop") or (asm.type=="int3" and nextAsm.type == "int3"): + if not asm.registersTouch(nextAsm): + toPatch = nextAsm.rawBytes + asm.rawBytes + outflankPatch = OutflankPatch( + idx, + asm.offset, + b"\x89\xc0", # mov eax, eax + asm, + nextAsm, + "Replace NOP".format(), + "." + ) + results.append(outflankPatch) + n += 1 # skip nextAsm - if 'cc int3' in s: - int3Lines.append(PossiblePatch(line.offset, idx)) + if (asm.type == "mov" and nextAsm.type == "mov") or (asm.type=="lea" and nextAsm.type == "lea"): + if not asm.registersTouch(nextAsm): + toPatch = nextAsm.rawBytes + asm.rawBytes + outflankPatch = OutflankPatch( + idx, + asm.offset, + toPatch, + asm, + nextAsm, + "Swap {}".format(asm.type), + "." + ) + results.append(outflankPatch) + n += 1 # skip nextAsm - # check for double-nop (very reliable) - for idx, possibleMatch in enumerate(nopLines): - # check if next byte is a nop too - if (idx+1 < len(nopLines)) and (possibleMatch.offset + 1 == nopLines[idx+1].offset): - # $ rasm2 -a x86 -b 64 -d '89c0' - # mov eax, eax - outflankPatch = OutflankPatch( - possibleMatch.matchId, - possibleMatch.offset, - b"\x89\xc0", - "Replace NOP;NOP with mov eax,eax", - "No side effects" - ) - results.append(outflankPatch) - - # double nop is enough - #if len(results) > 0: - # return results - - # check for int3 - for idx, possibleMatch in enumerate(int3Lines): - outflankPatch = OutflankPatch( - possibleMatch.matchId, - possibleMatch.offset, - b"\x90", - "Replace int3 with NOP", - "No real side effects" - ) - results.append(outflankPatch) - - # int3 replace is fine - #if len(results) > 0: - # return results - - # find single-nops - for idx, possibleMatch in enumerate(nopLines): - outflankPatch = OutflankPatch( - possibleMatch.matchId, - possibleMatch.offset, - b"\xfc", - "Replace NOP with cld (clear direction flag)", - "Few side effects" - ) - results.append(outflankPatch) + n += 1 # scan results, remove one's which gets detected if scanner is None: return results ret = [] + data: Data = filePe.DataCopy() for patch in results: - data: Data = filePe.DataCopy() + print("Patch location {} with: {} -> {}".format(patch.offset, patch.replaceBytes, patch.info)) data.patchData(patch.offset, patch.replaceBytes) + ret.append(patch) if not scanner.scannerDetectsBytes(data.getBytes(), filePe.filename): - logging.warn("Outflank OK! " + str(patch)) + logging.warn("Outflank possible") + return ret + #else: + # logging.warn("Outflank failed: " + str(patch)) + + # fail + logging.info("Outflank failed with attempted {} patches".format(len(results))) + return [] - ret.append(patch) - else: - logging.warn("Outflank failed: " + str(patch)) - - return ret - - -# https://stackoverflow.com/questions/14693701/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python -def escape_ansi(line): - ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]') - return ansi_escape.sub('', line) diff --git a/tests/test_pe.py b/tests/test_pe.py index b65cac1..8dd8d50 100644 --- a/tests/test_pe.py +++ b/tests/test_pe.py @@ -10,7 +10,7 @@ from plugins.outflank_pe import outflankPe from tests.scanners import * from model.model import Match, OutflankPatch from model.testverify import MatchConclusion, VerifyStatus -from utils import hexdmp, hexstr +from utils import hexdmp, hexstr, removeAnsi import r2pipe @@ -108,13 +108,16 @@ class PeTest(unittest.TestCase): start = 2807 # AF7 size = 8 matchAsmInstructions, matchDisasmLines = disassemble( - r2, filePe, start, size, moreUiLines=False) + r2, filePe, start, size, moreUiLines=0) + + #for a in matchDisasmLines: + # print(a) self.assertEqual(start, matchDisasmLines[0].offset) self.assertEqual(0x004014f7, matchDisasmLines[0].rva) - self.assertTrue('nop' in matchDisasmLines[0].text) - self.assertTrue('add rsp, 0x28' in matchDisasmLines[1].text) - self.assertTrue('ret' in matchDisasmLines[2].text) + self.assertTrue('nop' in removeAnsi(matchDisasmLines[0].text)) + self.assertTrue('add rsp, 0x28' in removeAnsi(matchDisasmLines[1].text)) + self.assertTrue('ret' in removeAnsi(matchDisasmLines[2].text)) self.assertEqual(start, matchAsmInstructions[0].offset) self.assertEqual(0x004014f7, matchAsmInstructions[0].rva) @@ -136,42 +139,58 @@ class PeTest(unittest.TestCase): start = 2807 # AF7 size = 8 matchAsmInstructions, matchDisasmLines = disassemble( - r2, filePe, start, size, moreUiLines=False) + r2, filePe, start, size, moreUiLines=0) # 0x004014f7 90 nop # 0x004014f8 4883c428 add rsp, 0x28 # 0x004014fc c3 ret - self.assertTrue('nop' in matchDisasmLines[0].text) - self.assertTrue('add rsp, 0x28' in matchDisasmLines[1].text) - self.assertTrue('ret' in matchDisasmLines[2].text) + self.assertTrue('nop' in removeAnsi(matchDisasmLines[0].text)) + self.assertTrue('add rsp, 0x28' in removeAnsi(matchDisasmLines[1].text)) + self.assertTrue('ret' in removeAnsi(matchDisasmLines[2].text)) filePe.data.swapData(2807, 1, 2807+1, 4) self.assertEqual(filePe.data.getBytesRange(2807, 2807+4), b"\x48\x83\xc4\x28") self.assertEqual(filePe.data.getBytesRange(2807+4, 2807+4+1), b"\x90") #matchAsmInstructions, matchDisasmLines = disassemble( - # r2, filePe, start, size, moreUiLines=False) + # r2, filePe, start, size, moreUiLines=0) #self.assertEqual('nop', matchAsmInstructions[1].disasm) #self.assertEqual('add rsp, 0x28', matchAsmInstructions[0].disasm) #self.assertEqual('ret', matchAsmInstructions[2].disasm) - def test_pe_outflank(self): + def test_disasm_outflank(self): filePe = FilePe() - filePe.loadFromFile("tests/data/test.exe") + filePe.loadFromFile("tests/data/test.exe") + r2 = r2pipe.open(filePe.filepath) + r2.cmd("aaa") + + fileOffset = filePe.codeRvaToOffset(0x0040154e) + matchAsmInstructions, _ = disassemble( + r2, filePe, fileOffset, 5, moreUiLines=0) + + # 0x0040154e 48894dd0 mov qword [var_30h], rcx ; format + # 0x00401552 488955d8 mov qword [var_28h], rdx ; arg2 + # 0x00401556 4c8945e0 mov qword [var_20h], r8 ; arg3 + # 0x0040155a 4c894de8 mov qword [var_18h], r9 ; arg4 + #for entry in matchDisasmLines: + # print(entry) + + self.assertEqual(len(matchAsmInstructions), 5) + + #for entry in matchAsmInstructions: + # print(entry) - # 0 0x00000600 0x6c00 0x00401000 0x7000 -r-x .text matches = [] - match = Match(0, 0x600 + 0x4d0, 8) # 8 is another NOP + match = Match(0, fileOffset, 8) matches.append(match) - - # the match should be good verifyStatus = [ VerifyStatus.GOOD ] matchConclusion = MatchConclusion(verifyStatus) - augmentFilePe(filePe, matches) - patches = outflankPe(filePe, matches, matchConclusion) - self.assertEqual(3, len(patches)) - self.assertEqual(2774, patches[0].offset) - self.assertEqual(2, len(patches[0].replaceBytes)) + #for patch in patches: + # print(patch) + + self.assertEqual(patches[0].offset, 2894) + self.assertEqual(patches[1].offset, 2902) + self.assertEqual(patches[2].offset, 2914) diff --git a/utils.py b/utils.py index 7712775..18923ae 100644 --- a/utils.py +++ b/utils.py @@ -2,6 +2,7 @@ import logging import json import os import base64 +import re from model.model import Match from model.testverify import FillType @@ -63,3 +64,9 @@ def hexstr(src: bytes, offset=0, length=0): byte_buffer = src[offset:offset+length] hex_string = ' '.join([f'{x:02x}' for x in byte_buffer]) return hex_string + + +# https://stackoverflow.com/questions/14693701/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python +def removeAnsi(line): + ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]') + return ansi_escape.sub('', line)