From 83edee2dfebafa6d3d4d762fdecb0549ce27f1e1 Mon Sep 17 00:00:00 2001 From: Dobin Rutishauser Date: Wed, 18 May 2022 13:54:21 +0200 Subject: [PATCH] my own implementation --- avred.py | 6 ++- dobin.py | 115 +++++++++++++++++++++++++++++++++++++++++++++++++++++ find.py | 22 ++++++---- scanner.py | 1 - test.py | 43 ++++++++++++++++---- todo.txt | 44 ++++++++++++++++++-- 6 files changed, 210 insertions(+), 21 deletions(-) create mode 100644 dobin.py diff --git a/avred.py b/avred.py index 51b3b88..47b0a5b 100644 --- a/avred.py +++ b/avred.py @@ -21,11 +21,13 @@ rootLogger.addHandler(consoleHandler) def test(): #pe, matches = test1() #pe, matches = test2() - pe, matches = test3() + #pe, matches = test3() + pe, matches = test4() for match in matches: for i in sorted(match): - print(f"[*] Signature between {i.begin} and {i.end}: ") + size = i.end - i.begin + print(f"[*] Signature between {i.begin} and {i.end} size {size}: ") data = pe.data[i.begin:i.end] print(hexdump.hexdump(data, result='return')) diff --git a/dobin.py b/dobin.py new file mode 100644 index 0000000..42b5560 --- /dev/null +++ b/dobin.py @@ -0,0 +1,115 @@ +from pe_utils import * +from find import bytes_detection + + +""" +attempts to locate the part in a PE file that causes the antivirus detection +""" +def findDetectedSections(pe, scanner): + detected_sections = [] + + for section in pe.sections: + new_pe = deepcopy(pe) + hide_section(new_pe, section.name) + + status = scanner.scan(new_pe.data) + section.detected = status + + if not status: + logging.info(f"Section {section.name} triggers the antivirus") + detected_sections += [section] + + sectionCount = len(detected_sections) + print(f"{sectionCount} section(s) trigger the antivirus") + for section in detected_sections: + print(f" section: {section.name}") + + return detected_sections + + +def investigate(pe, scanner): + detected = scanner.scan(pe.data) + if not detected: + logging.error(f"{pe.filename} is not detected by {scanner.scanner_name}") + return + + # identify which sections get detected + detected_sections = findDetectedSections(pe, scanner) + + # analyze each section + matches = [] + for section in detected_sections: + logging.info(f"Launching bytes analysis on section {section.name}: {section.addr}-{section.addr+section.size}") + match = _detection(scanner, pe.data, section.addr, section.addr+section.size) + matches.append(match) + + return matches + + +def makePatchedFile(fileData, offset, size): + patch = bytes(chr(0),'ascii') * int(size) + goat = fileData[:offset] + patch + fileData[offset+size:] + return goat + + +SIG_SIZE = 128 + +def _detection(scanner, fileData, sectionStart, sectionEnd): + size = sectionEnd - sectionStart + chunkSize = int(size // 2) + + logging.info(f"Testing: {sectionStart}-{sectionEnd} with size {sectionEnd-sectionStart} (chunkSize {chunkSize} bytes)") + #logging.info(f"Testing Top: {sectionStart}-{sectionStart+chunkSize} (chunkSize {chunkSize} bytes)") + #logging.info(f"Testing Bot: {sectionStart+chunkSize}-{sectionStart+chunkSize+chunkSize} (chunkSize {chunkSize} bytes)") + + if chunkSize < 2: + logging.error(f"Very small chunksize for a signature, problem?") + return [] + + chunkTopNull = makePatchedFile(fileData, sectionStart, chunkSize) + chunkBotNull = makePatchedFile(fileData, sectionStart+chunkSize, chunkSize) + + detectTopNull = scanner.scan(chunkTopNull) + detectBotNull = scanner.scan(chunkBotNull) + + if detectTopNull and detectBotNull: + # Both halves are detected + # Continue scanning both halves independantly, but with each other halve + # zeroed out (instead of the complete file) + logging.error("Both halves are detected!") + ret = [] + _detection(scanner, chunkBotNull, sectionStart, sectionStart+chunkSize) + _detection(scanner, chunkTopNull, sectionStart+chunkSize, sectionEnd) + + elif not detectTopNull and not detectBotNull: + if chunkSize < SIG_SIZE: + # No more detections + logging.info("No more detection") + logging.info(f"Result: {sectionStart}-{sectionEnd} ({sectionEnd-sectionStart} bytes)") + + print("Result:") + data = fileData[sectionStart:sectionStart+size] + print(hexdump.hexdump(data, result='return')) + else: + logging.info("No detections anymore, but too big. Continue anyway...") + _detection(scanner, fileData, sectionStart, sectionStart+chunkSize) + _detection(scanner, fileData, sectionStart+chunkSize, sectionEnd) + + #print("TopNull:") + #data = chunkBotNull[sectionStart:sectionStart+chunkSize] + #print(hexdump.hexdump(data, result='return')) + + #print("BotNull:") + #data = chunkTopNull[sectionStart+chunkSize:sectionStart+chunkSize+chunkSize] + #print(hexdump.hexdump(data, result='return')) + + elif not detectTopNull: + # Detection in the top half + #logging.info("Do Top") + return _detection(scanner, fileData, sectionStart, sectionStart+chunkSize) + elif not detectBotNull: + # Detection in the bottom half + #logging.info(f"Do Bot") + return _detection(scanner, fileData, sectionStart+chunkSize, sectionEnd) + + return [] \ No newline at end of file diff --git a/find.py b/find.py index 2aed788..946b2fc 100644 --- a/find.py +++ b/find.py @@ -16,10 +16,9 @@ todo: """ -#ResultQueue = deque() interval_tree = IntervalTree() START_LEAP = 2048 -MIN_LEAP = 8 +MIN_LEAP = 4 # 4 is like the minimum to work MAX_THREADS = 10 @@ -59,24 +58,33 @@ def filter_matches(good_res): the detection verdict. If it does, the half is added to a queue in order to improve the precision. Problem if each half is detected for now + + buffer: complete file + ResultQueue: to store results + current_offset: where in the buffer we are (initially begin of section) + end: offset end of the section + counter: just informative + scannner: which scanner we use """ def sigseek(buffer, ResultQueue, current_offset, end, counter, scanner): leap = (end - current_offset) // 2 - patch = bytes(chr(0),'ascii')*int(leap) + patch = bytes(chr(0),'ascii') * int(leap) nb_chunk = (end - current_offset) // leap if not leap == 0 else 0 detected_chunks = 0 bufs = [] + # nb_chunk will always be 2 + #logging.info(f"\t\t[+] {nb_chunk} chunks to process") while current_offset < end and leap >= MIN_LEAP: #progress.set_postfix(current_offset=current_offset+leap, refresh=True) - logging.info(f"[-] Patching buffer of size = {len(buffer)}, offset = {current_offset}, leap = {leap}") + logging.info(f"[-] Patching buffer {len(buffer)} {counter}: offset={current_offset} leap={leap}") goat = buffer[:current_offset] + patch + buffer[current_offset+leap:] bufs += [goat] if not scanner.scan(goat): - has_lead = True + #has_lead = True logging.info(f"[+] Found signature between {current_offset} and {current_offset+leap}") ResultQueue.append(Interval(int(current_offset), current_offset+leap)) else: @@ -107,7 +115,7 @@ def process_file(data, scanner, start = 0, end=-1): with futures.ThreadPoolExecutor(max_workers=MAX_THREADS) as exec: to_do = [] - last_size = 0 + #last_size = 0 while len(ResultQueue) > 0 or len(to_do) > 0: if len(ResultQueue) == 0: @@ -126,7 +134,7 @@ def process_file(data, scanner, start = 0, end=-1): else: return - last_size = match.end - match.begin + #last_size = match.end - match.begin futur = exec.submit(sigseek, data, ResultQueue, match.begin, match.end, counter, scanner) to_do = [futur] diff --git a/scanner.py b/scanner.py index 85f7422..1025ba7 100644 --- a/scanner.py +++ b/scanner.py @@ -39,7 +39,6 @@ class ScannerTestWeighted(Scanner): def scan(self, data): # 2/3 n = 0 - for detection in self.detections: fileData = data[detection.refPos:detection.refPos+len(detection.refData)] if fileData == detection.refData: diff --git a/test.py b/test.py index 07e8633..2e54b79 100644 --- a/test.py +++ b/test.py @@ -1,8 +1,10 @@ from scanner import ScannerTest, ScannerTestWeighted -from analyzer import * from pe_utils import * +#from analyzer import * +from analyzer import parse_pe +from dobin import investigate class TestDetection(): def __init__(self, refPos, refData): @@ -10,24 +12,31 @@ class TestDetection(): self.refData = refData def test1(): + # simple filename = "files/test.exe" detections = [] # one string in .rodata - detections.append( TestDetection(29824, b"Unknown error") ) + #detections.append( TestDetection(29824, b"Unknown error") ) + + # TODO PROBLEM with this one + #detections.append( TestDetection(30810, b"\xff\xff\x10\xb1\xff\xff\xc2\xb2\xff\xff") ) + + # WORKS + detections.append( TestDetection(30823, b"\xff\x98\xb0\xff\xff\xdb\xb1\xff") ) + scanner = ScannerTest(detections) pe = parse_pe(filename) - matches = investigate(pe, scanner) return pe, matches def test2(): + # two sections filename = "files/test.exe" detections = [] # .rodata detections.append( TestDetection(29824, b"Unknown error") ) - # .text detections.append( TestDetection(1664, b"\xf4\x63\x00\x00\xe8\x87\x6a\x00\x00\x48\x8b\x15\x40") ) scanner = ScannerTest(detections) @@ -38,16 +47,34 @@ def test2(): def test3(): + # two in one section filename = "files/test.exe" detections = [] # .rodata detections.append( TestDetection(29824, b"Unknown error") ) detections.append( TestDetection(31850, b" 10.2.0") ) - # .text - detections.append( TestDetection(1664, b"\xf4\x63\x00\x00\xe8\x87\x6a\x00\x00\x48\x8b\x15\x40") ) - scanner = ScannerTestWeighted(detections) - + #detections.append( TestDetection(1664, b"\xf4\x63\x00\x00\xe8\x87\x6a\x00\x00\x48\x8b\x15\x40") ) + + scanner = ScannerTest(detections) + pe = parse_pe(filename) + matches = investigate(pe, scanner) + return pe, matches + + +def test4(): + # weighted + filename = "files/test.exe" + detections = [] + # .rodata + detections.append( TestDetection(29824, b"Unknown error") ) + detections.append( TestDetection(30823, b"\xff\x98\xb0\xff\xff\xdb\xb1\xff") ) + detections.append( TestDetection(31850, b" 10.2.0") ) + + # .text + #detections.append( TestDetection(1664, b"\xf4\x63\x00\x00\xe8\x87\x6a\x00\x00\x48\x8b\x15\x40") ) + + scanner = ScannerTestWeighted(detections) pe = parse_pe(filename) matches = investigate(pe, scanner) return pe, matches diff --git a/todo.txt b/todo.txt index ebfb05e..e001845 100644 --- a/todo.txt +++ b/todo.txt @@ -1,5 +1,43 @@ -* MIN_LEAP = 4 - * merge adjectant matches - * remove duplicate matches ++ MIN_LEAP = 4 + + merge adjectant matches + + remove duplicate matches +* problems + * if each half is detected + * add test + * if detection is score based + + +------- + +offset slightly wrong generated bad results, how to minimize? + + +[INFO ][2022-05-18 12:38:43,356][dobin.py: 42] investigate() :: Launching bytes analysis on section .rdata: 29696-33280 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29696-33280: 1792 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 95] _detection() :: Do Top +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29696-31488: 896 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 95] _detection() :: Do Top +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29696-30592: 448 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 95] _detection() :: Do Top +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29696-30144: 224 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 95] _detection() :: Do Top +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29696-29920: 112 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 98] _detection() :: Do Bot +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29808-29920: 56 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 95] _detection() :: Do Top +[INFO ][2022-05-18 12:38:43,356][dobin.py: 59] _detection() :: 29808-29864: 28 +[INFO ][2022-05-18 12:38:43,356][dobin.py: 78] _detection() :: No more detection +[INFO ][2022-05-18 12:38:43,356][dobin.py: 79] _detection() :: Result: 29808-29864 size:56 +Result: +00000000: 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ +00000010: 55 6E 6B 6E 6F 77 6E 20 65 72 72 6F 72 00 00 00 Unknown error... +00000020: 41 72 67 75 6D 65 6E 74 20 64 6F 6D 61 69 6E 20 Argument domain +00000030: 65 72 72 6F 72 20 28 44 error (D +TopNull: +00000000: 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ +00000010: 55 6E 6B 6E 6F 77 6E 20 65 72 72 6F Unknown erro +BotNull: +00000000: 72 00 00 00 41 72 67 75 6D 65 6E 74 20 64 6F 6D r...Argument dom +00000010: 61 69 6E 20 65 72 72 6F 72 20 28 44 ain error (D \ No newline at end of file