mirror of
https://github.com/volatilityfoundation/volatility3.git
synced 2026-08-24 07:02:23 +02:00
Conver the individual tests to the scanner framework, and work that into the automagic.
This commit is contained in:
@@ -1,17 +1,14 @@
|
||||
from volatility.framework import automagic
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.interfaces import configuration as config_interface
|
||||
|
||||
if __name__ == "__main__":
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))))
|
||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
|
||||
|
||||
import struct
|
||||
|
||||
from volatility.framework import interfaces, layers, validity
|
||||
from volatility.framework.interfaces import automagic as automagic_interface
|
||||
from volatility.framework import automagic, interfaces, layers, validity
|
||||
from volatility.framework.configuration import requirements
|
||||
from volatility.framework.interfaces import automagic as automagic_interface, configuration as config_interface
|
||||
|
||||
PAGE_SIZE = 0x1000
|
||||
|
||||
@@ -49,21 +46,21 @@ class DtbTest(validity.ValidityRoutines):
|
||||
def unpack(self, value):
|
||||
return struct.unpack("<" + self.ptr_struct, value)[0]
|
||||
|
||||
def run(self, page_offset, ctx, layer_name):
|
||||
value = ctx.memory.read(layer_name, page_offset + (self.ptr_reference * self.ptr_size),
|
||||
self.ptr_size)
|
||||
def __call__(self, data, data_offset, page_offset):
|
||||
value = data[page_offset + (self.ptr_reference * self.ptr_size):page_offset + (
|
||||
(self.ptr_reference + 1) * self.ptr_size)]
|
||||
ptr = self.unpack(value)
|
||||
# The value *must* be present (bit 0) since it's a mapped page
|
||||
# It's almost always writable (bit 1)
|
||||
# It's occasionally Super, but not reliably so, haven't checked when/why not
|
||||
# The top 3-bits are usually ignore (which in practice means 0
|
||||
# Need to find out why the middle 3-bits are usually 6 (0110)
|
||||
if ptr != 0 and (ptr & self.mask == page_offset) & (ptr & 0xFF1 == 0x61):
|
||||
if ptr != 0 and (ptr & self.mask == data_offset + page_offset) & (ptr & 0xFF1 == 0x61):
|
||||
dtb = (ptr & self.mask)
|
||||
return self.second_pass(dtb, ctx, layer_name)
|
||||
return self.second_pass(dtb, data, data_offset)
|
||||
|
||||
def second_pass(self, dtb, ctx, layer_name):
|
||||
data = ctx.memory.read(layer_name, dtb, PAGE_SIZE)
|
||||
def second_pass(self, dtb, data, data_offset):
|
||||
page = data[dtb - data_offset:data - data_offset + PAGE_SIZE]
|
||||
usr_count, sup_count = 0, 0
|
||||
for i in range(0, PAGE_SIZE, self.ptr_size):
|
||||
val = self.unpack(data[i:i + self.ptr_size])
|
||||
@@ -73,7 +70,7 @@ class DtbTest(validity.ValidityRoutines):
|
||||
# print(hex(dtb), usr_count, sup_count, usr_count + sup_count)
|
||||
# We sometimes find bogus DTBs at 0x16000 with a very low sup_count and 0 usr_count
|
||||
if usr_count or sup_count > 5:
|
||||
return (usr_count, -sup_count), dtb
|
||||
return dtb
|
||||
|
||||
|
||||
class DtbTest32bit(DtbTest):
|
||||
@@ -105,12 +102,14 @@ class DtbTestPae(DtbTest):
|
||||
ptr_reference = 0x3,
|
||||
mask = 0x3FFFFFFFFFF000)
|
||||
|
||||
def second_pass(self, dtb, ctx, layer_name):
|
||||
def second_pass(self, dtb, data, data_offset):
|
||||
dtb -= 0x4000
|
||||
data = ctx.memory.read(layer_name, dtb, PAGE_SIZE)
|
||||
val = self.unpack(data[3 * self.ptr_size: 4 * self.ptr_size])
|
||||
if (val & self.mask == dtb + 0x4000) and (val & 0xFFF == 0x001):
|
||||
return val, dtb
|
||||
# If we're not in something that the overlap would pick up
|
||||
if dtb - data_offset >= 0:
|
||||
pointers = data[dtb - data_offset + (3 * self.ptr_size): dtb - data_offset + (4 * self.ptr_size)]
|
||||
val = self.unpack(pointers)
|
||||
if (val & self.mask == dtb + 0x4000) and (val & 0xFFF == 0x001):
|
||||
return dtb
|
||||
|
||||
|
||||
class SelfReferentialTest(object):
|
||||
@@ -132,8 +131,25 @@ class SelfReferentialTest(object):
|
||||
|
||||
|
||||
class PageMapScanner(interfaces.layers.ScannerInterface):
|
||||
overlap = 0x4000
|
||||
tests = [DtbTest32bit, DtbTest64bit, DtbTestPae]
|
||||
|
||||
def __init__(self, tests):
|
||||
interfaces.layers.ScannerInterface.__init__(self)
|
||||
for value in tests:
|
||||
self._check_type(value, DtbTest)
|
||||
self.tests = tests
|
||||
|
||||
def __call__(self, data, data_offset):
|
||||
pass
|
||||
results = {}
|
||||
for test in self.tests:
|
||||
results[test] = set()
|
||||
|
||||
for test in self.tests:
|
||||
for page_offset in range(0, len(data), PAGE_SIZE):
|
||||
result = test(data, data_offset, page_offset)
|
||||
if result is not None:
|
||||
yield (test, result)
|
||||
|
||||
|
||||
class PageMapOffsetHelper(automagic_interface.AutomagicInterface):
|
||||
@@ -173,14 +189,11 @@ class PageMapOffsetHelper(automagic_interface.AutomagicInterface):
|
||||
requirement.requirements["memory_layer"].validate(context, sub_config_path)):
|
||||
physical_layer = requirement.requirements["memory_layer"].config_value(context, sub_config_path)
|
||||
# TODO: Convert to scanner framework
|
||||
# context.memory[physical_layer].scan(context, scanner)
|
||||
hits = scan(context, physical_layer, useful)
|
||||
# TODO: At this point, we know the class, so ditch the other tests
|
||||
for test in useful:
|
||||
if hits.get(test.layer_type, []):
|
||||
context.config[config_interface.path_join(sub_config_path, "page_map_offset")] = \
|
||||
hits[test.layer_type][0]
|
||||
requirement.construct(context, config_path)
|
||||
hits = context.memory[physical_layer].scan(context, PageMapScanner(useful))
|
||||
for test, dtb in hits:
|
||||
context.config[config_interface.path_join(sub_config_path, "page_map_offset")] = dtb
|
||||
requirement.construct(context, config_path)
|
||||
break
|
||||
else:
|
||||
for subreq in requirement.requirements.values():
|
||||
self(context, subreq, sub_config_path)
|
||||
@@ -223,17 +236,17 @@ if __name__ == '__main__':
|
||||
if tests:
|
||||
for i in range(len(args.filenames)):
|
||||
print("[*] Scanning " + args.filenames[i] + "...")
|
||||
hits = scan(ctx, "data" + str(i), tests)
|
||||
for key in tests:
|
||||
arch_hits = hits.get(key.layer_type, [])
|
||||
if arch_hits:
|
||||
print(" ", key.layer_type.__name__ + ": " + repr([hex(x) for x in sorted(arch_hits)]))
|
||||
guesses = []
|
||||
for key in hits:
|
||||
guesses.append((len(hits[key]), key.__name__, hits[key]))
|
||||
num, arch, dtbs = max(guesses)
|
||||
if num:
|
||||
print("[!] OS Guess:", arch, "with DTB", hex(dtbs[0]))
|
||||
guesses = dict([(test.layer_type.__name__, set()) for test in tests])
|
||||
hits = ctx.memory["data" + str(i)].scan(ctx, PageMapScanner(tests))
|
||||
for test, dtb in sorted(hits):
|
||||
guesses[test.layer_type.__name__].add(dtb)
|
||||
arch = None
|
||||
for guess in sorted(guesses, key = lambda x: -len(guesses[x])):
|
||||
if not arch and len(guesses[guess]) > 0:
|
||||
arch = guess
|
||||
print(" " + guess + ": " + ", ".join([hex(x) for x in sorted(guesses[guess])]))
|
||||
if arch:
|
||||
print("[!] OS Guess:", arch, "with DTB", hex(list(guesses[arch])[0]))
|
||||
else:
|
||||
print("[X] No DTBs found")
|
||||
print()
|
||||
|
||||
Reference in New Issue
Block a user