Clean up, use a config file to avoid hardcoding paths

Author: plowsec

antivirus_debugger.py

@@ -1,20 +1,12 @@
-import logging
-import os
-import sys
-import random
-import base64
-import r2pipe
-import string
-import shutil
 import argparse
+import sys
+from tempfile import NamedTemporaryFile
 
-from copy import deepcopy
-from dataclasses import dataclass
-from scanner import DockerWindowsDefender, VMWareDeepInstinct, VMWareKaspersky, VMWareAvast
 from find import bytes_detection
 from find_bad_strings import bissect
 from pe_utils import *
-from tempfile import NamedTemporaryFile
+from scanner import DockerWindowsDefender
+from scanner import g_scanner
 
 log_format = '[%(levelname)-8s][%(asctime)s][%(filename)s:%(lineno)3d] %(funcName)s() :: %(message)s'
 logging.basicConfig(filename='debug.log',
@@ -32,7 +24,6 @@
 consoleHandler.setFormatter(logFormatter)
 rootLogger.addHandler(consoleHandler)
 
-g_scanner = None
 BINARY = ""
 
 g_args = None
@@ -136,7 +127,7 @@ def strings_analysis(pe):
     if detection_result:
         logging.info(f"{g_scanner.scanner_name} seems to only detect strings in this binary.")
     else:
-        logging.warn(f"Patching all the strings does not evade detection.")
+        logging.warning(f"Patching all the strings does not evade detection.")
 
     return detection_result
 
@@ -275,9 +266,6 @@ def parse_pe(sample_file):
 
 if __name__ == "__main__":
 
-    #g_scanner = VMWareAvast()
-    #g_scanner = VMWareDeepInstinct()
-    g_scanner = DockerWindowsDefender()
     parser = argparse.ArgumentParser()
 
     parser.add_argument("-s", '--skip-strings', help="Skip strings analysis", action="store_true")
@@ -288,7 +276,22 @@ def parse_pe(sample_file):
     parser.add_argument('-c', '--section', help="Analyze provided section")
     parser.add_argument('-g', '--globals', help="Analyze global variables in .data section", action="store_true")
     parser.add_argument('-V', '--virus', help="Virus scan", action="store_true")
+    parser.add_argument('-H', '--hide-section', help="Hide a section", type=str)
+    parser.add_argument('-S', "--scanner", help="Antivirus engine", default="DockerWindowsDefender")
     g_args = parser.parse_args()
 
+    if g_args.scanner == "DockerWindowsDefender":
+        g_scanner = DockerWindowsDefender()
+
     pe = parse_pe(g_args.file)
+
+    if g_args.hide_section:
+        copy_file = NamedTemporaryFile(delete=False)
+        shutil.copy(pe.filename, copy_file.name)
+        pe.filename = copy_file.name
+        hide_section(pe, g_args.hide_section)
+        logging.info(f"Dumped patched file @ {copy_file.name}")
+        sys.exit(0)
+
+
     investigate(pe)

config.json

@@ -0,0 +1,10 @@
+{
+  "loadlibrary_path": "/home/toto/loadlibrary/mpclient",
+  "loadlibrary_path_dir": "/home/toto/loadlibrary",
+  "docker_loadlibrary_name": "loadlibrary-working",
+  "avast_vmx": "/Users/vladimir/Virtual Machines.localized/Windows_10_Avast.vmwarevm/Windows_10_Avast.vmx",
+  "kaspersky_vmx": "../../kasp.vmx",
+  "deepinstinct_vmx": "../../Virtual Machines.localized/Windows_10_AV_DeepInstinct.vmwarevm/Windows_10_AV_DeepInstinct.vmx",
+  "vmware_user": "toto",
+  "vmware_passwd": "$crt1234!"
+}

config.py

@@ -0,0 +1,23 @@
+import json
+import os
+
+CONFIG_FILE = os.path.join(os.path.dirname(__file__), "config.json")
+
+
+def get_value(value):
+
+    try:
+        with open(CONFIG_FILE) as jsonfile:
+
+            data = json.load(jsonfile)
+
+        if value in data:
+            return data[value]
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        import os
+        print(os.getcwd())
+        raise Exception(e)
+
+    raise Exception(f"Unknown field: {value}. Available fields are: {data.keys()}")

find.py

@@ -1,19 +1,19 @@
 #!/usr/bin/python
 
-import os
-import subprocess
-import re
-import sys
-import hexdump
+import logging
 import string
-from tqdm import tqdm
-from concurrent import futures
+import sys
 from collections import deque
-from intervaltree import Interval, IntervalTree
+from concurrent import futures
+
+import hexdump
 from capstone import *
-from scanner import WindowsDefender, DockerWindowsDefender, VMWareDeepInstinct, VMWareKaspersky, VMWareAvast
-import logging
+from intervaltree import Interval, IntervalTree
+from tqdm import tqdm
+
+import config
 import pe_utils
+from scanner import g_scanner
 
 logging.basicConfig(filename='debug.log',
                     filemode='a',
@@ -36,7 +36,7 @@
 IGNORE_START = 0
 IGNORE_END = 0x256 #todo use pefile to find the start of code
 GOAT_FILE = '/tmp/metsrv.x64.dll'
-WDEFENDER_INSTALL_PATH = '/home/vladimir/tools/loadlibrary/'
+WDEFENDER_INSTALL_PATH = config.get_value("loadlibrary_path_dir")
 MAX_THREADS = 10
 DEBUG_LEVEL = 1
 buffer = []
@@ -47,7 +47,6 @@
 cs = Cs(CS_ARCH_X86, CS_MODE_64)
 cs.detail = True
 cs.skipdata = True
-g_scanner = None
 
 
 
@@ -296,9 +295,7 @@ def process_file(sample_file, start = 0, end=-1):
 
 
 def bytes_detection(filename, start=0, end=-1):
-    global g_scanner
 
-    g_scanner = DockerWindowsDefender()
     # g_scanner = VMWareAvast()
     sample_file = filename
 

find_bad_strings.py

@@ -1,32 +1,29 @@
 #!/usr/bin/python3
-import sys
-import string
-import os
-import subprocess
 import dataclasses
+import logging
 import re
-import random
+import subprocess
+import sys
 import tempfile
+
 from tqdm import tqdm
-from itertools import islice
-from scanner import WindowsDefender, DockerWindowsDefender, VMWareKaspersky, VMWareAvast
-import logging
+
+from scanner import g_scanner
 
 logging.basicConfig(filename='debug.log',
                     filemode='a',
                     format='[%(levelname)-8s][%(asctime)s][%(filename)s:%(lineno)3d] %(funcName)s() :: %(message)s',
                     datefmt='%Y/%m/%d %H:%M',
                     level=logging.DEBUG)
 
-BINARY                 = "/home/vladimir/dev/av-signatures-finder/test_cases/ext_server_kiwi.x64.dll"
+BINARY                 = "test_cases/ext_server_kiwi.x64.dll"
 ORIGINAL_BINARY        = ""
-WDEFENDER_INSTALL_PATH = '/home/vladimir/tools/new_loadlibrary/loadlibrary/'
 DEBUG_LEVEL            = 2  # setting supporting levels 0-3, incrementing the verbosity of log msgs
 LVL_ALL_DETAILS        = 3  # everything
 LVL_DETAILS            = 2  # only    important  details
 LVL_RES_ONLY           = 1  # only    results
 LVL_SILENT             = 0  # quiet
-g_scanner = None
+
 
 @dataclasses.dataclass
 class StringRef:
@@ -340,7 +337,12 @@ def rec_bissect(binary, string_refs, blacklist):
         return blacklist
 
 
-    half_nb_strings = len(string_refs) // 2
+    try:
+        half_nb_strings = len(string_refs) // 2
+    except:
+        print_dbg(f"Found it: f{repr(string_refs)}", LVL_RES_ONLY, False)
+        blacklist.append(string_refs.index)
+        return blacklist
     half1 = string_refs[:half_nb_strings]
     half2 = string_refs[half_nb_strings:]
     binary1 = binary
@@ -422,13 +424,10 @@ def rec_bissect(binary, string_refs, blacklist):
 """
 def bissect(sample_file, blacklist = []):
 
-    global g_scanner
     global BINARY
 
     BINARY = sample_file
-    if g_scanner is None:
-        g_scanner = DockerWindowsDefender()
-        # g_scanner = VMWareAvast()
+
     # no point in continuing if the binary is not detected as malicious already.
     assert(scan(sample_file) is True)
 
@@ -477,20 +476,26 @@ def bissect(sample_file, blacklist = []):
     if len(blacklist) > 0:
         print_dbg(f"Found {len(blacklist)} signatures", LVL_DETAILS, True)
         print(blacklist)
+
+        for b in blacklist:
+            string = next(filter(lambda x: x.index == b, str_refs))
+            logging.info(f"String @ {hex(string.paddr)} should be patched: ")
+            logging.info(string.content)
         tmpfile = "/tmp/newbin"
-        if not validate_results(ORIGINAL_BINARY, tmpfile, blacklist, str_refs):
-            print_dbg("Validation is ok !", LVL_DETAILS, True)
-        else:
-            print_dbg("Patched binary is still detected, retrying.", LVL_DETAILS, True)
-            bissect("/tmp/newbin", blacklist)
+        if ORIGINAL_BINARY != "":
+            if not validate_results(ORIGINAL_BINARY, tmpfile, blacklist, str_refs):
+                print_dbg("Validation is ok !", LVL_DETAILS, True)
+            else:
+                print_dbg("Patched binary is still detected, retrying.", LVL_DETAILS, True)
+                bissect("/tmp/newbin", blacklist)
     else:
         print_dbg("No signatures found...", LVL_DETAILS, True)
     return blacklist
 
 
 if __name__ == "__main__":
 
-    g_scanner = DockerWindowsDefender()
+    #g_scanner = DockerWindowsDefender()
 
 
     sample_file = BINARY

pe_utils.py

@@ -221,17 +221,23 @@ def parse_strings_old(strings_data):
     return string_refs
 
 
-def parse_strings(filename, all_sections=False, min_length=5):
+def parse_strings(filename, all_sections=False, min_length=12):
     pipe = r2pipe.open(filename)
     pipe.cmd("aaa")
     # pipe.cmd("aaa")
-    strings = pipe.cmdj("izzj")
+    if all_sections:
+        strings = pipe.cmdj("izzj")
+    else:
+        strings = pipe.cmdj("izj")
 
     string_refs = []
 
     for string in strings:
 
-        if string.get("size") < min_length:
+        if string.get("length") < min_length:
+            continue
+
+        if not string.get("section").startswith("."):
             continue
 
         str_ref = StringRef()

scan.py

@@ -0,0 +1,30 @@
+import os
+import subprocess
+import sys
+import re
+import config
+
+WDEFENDER_INSTALL_PATH = config.get_value("loadlibrary_path")
+
+current_dir = os.getcwd()
+os.chdir(os.path.dirname(WDEFENDER_INSTALL_PATH))
+command = [WDEFENDER_INSTALL_PATH, sys.argv[1]]
+p = subprocess.Popen(command, stdout=subprocess.PIPE,
+                     stderr=subprocess.STDOUT)
+
+while (True):
+
+    retcode = p.poll()  # returns None while subprocess is running
+    out = p.stdout.readline().decode('utf-8', errors='ignore')
+    print(out)
+    m = re.search('Threat', out)
+
+    if m:
+        print("Threat found\n")
+        exit(-1)
+
+    if (retcode is not None):
+        break
+
+os.chdir(current_dir)
+exit(0)