my own implementation

This commit is contained in:
Dobin Rutishauser
2022-05-18 13:54:21 +02:00
parent cd101f894b
commit 83edee2dfe
6 changed files with 210 additions and 21 deletions
+4 -2
View File
@@ -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'))
+115
View File
@@ -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 []
+15 -7
View File
@@ -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]
-1
View File
@@ -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:
+35 -8
View File
@@ -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
+41 -3
View File
@@ -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