mirror of
https://github.com/dobin/SuperMega
synced 2026-06-02 17:27:10 +00:00
refactor: log msgs, and var renaming
This commit is contained in:
+9
-2
@@ -103,8 +103,15 @@ class ExeHost():
|
||||
'type': reloc_type,
|
||||
})
|
||||
|
||||
# rwx
|
||||
self.rwx_section = pehelper.get_rwx_section(pe)
|
||||
# rwx section
|
||||
entrypoint = pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
for section in pe.sections:
|
||||
if (section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_READ'] and
|
||||
section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_WRITE'] and
|
||||
section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']
|
||||
):
|
||||
if entrypoint > section.VirtualAddress and entrypoint < section.VirtualAddress + section.Misc_VirtualSize:
|
||||
self.rwx_section = section
|
||||
|
||||
# If the PE file was loaded using the fast_load=True argument, we will need to parse the data directories:
|
||||
#pe.parse_data_directories()
|
||||
|
||||
+3
-16
@@ -10,7 +10,7 @@ from model.defs import *
|
||||
logger = logging.getLogger("PEHelper")
|
||||
|
||||
|
||||
def extract_code_from_exe(exe_file: FilePath) -> bytes:
|
||||
def extract_code_from_exe_file(exe_file: FilePath) -> bytes:
|
||||
pe = pefile.PE(exe_file)
|
||||
section = get_code_section(pe)
|
||||
data: bytes = section.get_data()
|
||||
@@ -39,24 +39,11 @@ def get_code_section(pe: pefile.PE) -> pefile.SectionStructure:
|
||||
raise Exception("Code section not found")
|
||||
|
||||
|
||||
# RWX
|
||||
def get_rwx_section(pe: pefile.PE) -> pefile.SectionStructure:
|
||||
entrypoint = pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
for section in pe.sections:
|
||||
if (section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_READ'] and
|
||||
section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_WRITE'] and
|
||||
section.Characteristics & pefile.SECTION_CHARACTERISTICS['IMAGE_SCN_MEM_EXECUTE']
|
||||
):
|
||||
if entrypoint > section.VirtualAddress and entrypoint < section.VirtualAddress + section.Misc_VirtualSize:
|
||||
return section
|
||||
return None
|
||||
|
||||
|
||||
# keystone/capstone stuff
|
||||
|
||||
def assemble_lea(current_address: int, destination_address: int, reg: str) -> bytes:
|
||||
print("LEAH: 0x{:X} - 0x{:X} = 0x{:X}".format(
|
||||
current_address, destination_address, destination_address - current_address))
|
||||
#print("LEAH: 0x{:X} - 0x{:X} = 0x{:X}".format(
|
||||
# current_address, destination_address, destination_address - current_address))
|
||||
offset = destination_address - current_address
|
||||
ks = Ks(KS_ARCH_X86, KS_MODE_64)
|
||||
encoding, _ = ks.asm(f"lea {reg}, qword ptr ds:[{offset}]")
|
||||
|
||||
+1
-1
@@ -21,7 +21,7 @@ def asm_to_shellcode(asm_in: FilePath, build_exe: FilePath, shellcode_out: FileP
|
||||
])
|
||||
if not os.path.isfile(build_exe):
|
||||
raise Exception("Compiling failed")
|
||||
code = extract_code_from_exe(build_exe)
|
||||
code = extract_code_from_exe_file(build_exe)
|
||||
observer.add_code("carrier_shc", code)
|
||||
with open(shellcode_out, 'wb') as f:
|
||||
f.write(code)
|
||||
|
||||
+8
-10
@@ -47,7 +47,7 @@ def inject_exe(
|
||||
# verify and log
|
||||
shellcode = file_readall_binary(shellcode_in)
|
||||
shellcode_len = len(shellcode)
|
||||
code = extract_code_from_exe(exe_out)
|
||||
code = extract_code_from_exe_file(exe_out)
|
||||
in_code = code[peinj.shellcodeOffsetRel:peinj.shellcodeOffsetRel+shellcode_len]
|
||||
jmp_code = code[peinj.backdoorOffsetRel:peinj.backdoorOffsetRel+12]
|
||||
if config.debug:
|
||||
@@ -58,11 +58,8 @@ def inject_exe(
|
||||
|
||||
|
||||
def injected_fix_iat(exe_out: FilePath, carrier: Carrier, exe_host: ExeHost):
|
||||
"""replace IAT in shellcode in code and re-implant it"""
|
||||
|
||||
# get code section of exe_out
|
||||
code = extract_code_from_exe(exe_out)
|
||||
|
||||
"""replace IAT-placeholders in shellcode with call's to the IAT"""
|
||||
code = extract_code_from_exe_file(exe_out)
|
||||
for iatRequest in carrier.get_all_iat_requests():
|
||||
if not iatRequest.placeholder in code:
|
||||
raise Exception("IatResolve ID {} not found, abort".format(iatRequest.placeholder))
|
||||
@@ -85,6 +82,7 @@ def injected_fix_iat(exe_out: FilePath, carrier: Carrier, exe_host: ExeHost):
|
||||
|
||||
|
||||
def injected_fix_data(exe_path, carrier: Carrier, exe_host: ExeHost):
|
||||
"""Inject shellcode-data into .rdata and replace reusedata_fixup placeholders in code with LEA"""
|
||||
# Insert my data into the .rdata section.
|
||||
# Chose and save each datareuse_fixup's addres.
|
||||
reusedata_fixups: List[DataReuseEntry] = carrier.get_all_reusedata_fixups()
|
||||
@@ -93,8 +91,8 @@ def injected_fix_data(exe_path, carrier: Carrier, exe_host: ExeHost):
|
||||
with open(exe_path, "r+b") as f:
|
||||
for datareuse_fixup in reusedata_fixups:
|
||||
var_data = datareuse_fixup.data
|
||||
print(" Addr: {} / 0x{:X} Data: {}".format(
|
||||
addr, addr, len(var_data)))
|
||||
#print(" Addr: {} / 0x{:X} Data: {}".format(
|
||||
# addr, addr, len(var_data)))
|
||||
f.seek(addr)
|
||||
f.write(var_data)
|
||||
datareuse_fixup.addr = addr + sect.virt_addr + exe_host.image_base - sect.raw_addr
|
||||
@@ -102,7 +100,7 @@ def injected_fix_data(exe_path, carrier: Carrier, exe_host: ExeHost):
|
||||
|
||||
# patch code section
|
||||
# replace the placeholder with a LEA instruction to the data we written above
|
||||
code = extract_code_from_exe(exe_path)
|
||||
code = extract_code_from_exe_file(exe_path)
|
||||
for datareuse_fixup in reusedata_fixups:
|
||||
if not datareuse_fixup.randbytes in code:
|
||||
raise Exception("DataResuse: ID {} not found, abort".format(
|
||||
@@ -111,7 +109,7 @@ def injected_fix_data(exe_path, carrier: Carrier, exe_host: ExeHost):
|
||||
offset_from_datasection = code.index(datareuse_fixup.randbytes)
|
||||
instruction_virtual_address = offset_from_datasection + exe_host.image_base + exe_host.code_virtaddr
|
||||
destination_virtual_address = datareuse_fixup.addr
|
||||
logger.info(" Replace {} at VA 0x{:x} with call to IAT at VA 0x{:x}".format(
|
||||
logger.info(" Replace {} at VA 0x{:x} with .rdata LEA at VA 0x{:x}".format(
|
||||
datareuse_fixup.randbytes, instruction_virtual_address, destination_virtual_address
|
||||
))
|
||||
lea = assemble_lea(
|
||||
|
||||
+4
-4
@@ -16,7 +16,7 @@ import phases.compiler
|
||||
import phases.assembler
|
||||
import phases.injector
|
||||
from observer import observer
|
||||
from peparser.pehelper import extract_code_from_exe
|
||||
from peparser.pehelper import extract_code_from_exe_file
|
||||
|
||||
from model.project import Project
|
||||
from model.settings import Settings
|
||||
@@ -266,14 +266,14 @@ def start(settings: Settings):
|
||||
project.carrier,
|
||||
project.exe_host)
|
||||
|
||||
code = extract_code_from_exe(settings.inject_exe_out)
|
||||
code = extract_code_from_exe_file(settings.inject_exe_out)
|
||||
pe = pefile.PE(settings.inject_exe_out)
|
||||
ep = pe.OPTIONAL_HEADER.AddressOfEntryPoint
|
||||
ep_raw = get_physical_address(pe, ep)
|
||||
pe.close()
|
||||
|
||||
print("Raw: {} / 0x{:x}".format(
|
||||
ep_raw, ep_raw))
|
||||
#print("Raw: {} / 0x{:x}".format(
|
||||
# ep_raw, ep_raw))
|
||||
observer.add_code("exe_fucking_final",
|
||||
code[ep_raw:ep_raw+300])
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ import logging
|
||||
|
||||
from model.exehost import ExeHost
|
||||
from model.defs import *
|
||||
from peparser.pehelper import extract_code_from_exe
|
||||
from peparser.pehelper import extract_code_from_exe_file
|
||||
from helper import hexdump
|
||||
from observer import observer
|
||||
|
||||
@@ -48,7 +48,7 @@ class DerBackdoorerTest(unittest.TestCase):
|
||||
)
|
||||
|
||||
self.assertTrue(result)
|
||||
code = extract_code_from_exe(exe_out_path)
|
||||
code = extract_code_from_exe_file(exe_out_path)
|
||||
extracted_code = code[peinj.shellcodeOffsetRel:peinj.shellcodeOffsetRel+len(shellcode)]
|
||||
self.assertEqual(shellcode, extracted_code)
|
||||
|
||||
@@ -80,7 +80,7 @@ class DerBackdoorerTest(unittest.TestCase):
|
||||
self.assertTrue(result)
|
||||
|
||||
# code
|
||||
code = extract_code_from_exe(exe_out_path)
|
||||
code = extract_code_from_exe_file(exe_out_path)
|
||||
extracted_code = code[peinj.shellcodeOffsetRel:peinj.shellcodeOffsetRel+len(shellcode)]
|
||||
self.assertEqual(shellcode, extracted_code)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user