Skip to content

Commit 5ff06fe

Browse files
authored
Merge pull request #1272 from volatilityfoundation/816-port-cmdscan-and-console-plugins-from-vol2-to-vol3-please
#816 cmdscan and console
2 parents 1409f47 + 1c3e557 commit 5ff06fe

15 files changed

+9056
-1
lines changed
Lines changed: 381 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,381 @@
1+
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
2+
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
3+
#
4+
5+
# This module attempts to locate windows console histories.
6+
7+
import logging
8+
import struct
9+
from typing import Tuple, Generator, Set, Dict, Any, Optional
10+
11+
from volatility3.framework import interfaces
12+
from volatility3.framework import renderers
13+
from volatility3.framework.configuration import requirements
14+
from volatility3.framework.layers import scanners
15+
from volatility3.framework.objects import utility
16+
from volatility3.framework.renderers import format_hints
17+
from volatility3.plugins.windows import pslist, consoles
18+
19+
20+
vollog = logging.getLogger(__name__)
21+
22+
23+
class CmdScan(interfaces.plugins.PluginInterface):
24+
"""Looks for Windows Command History lists"""
25+
26+
_required_framework_version = (2, 4, 0)
27+
_version = (1, 0, 0)
28+
29+
@classmethod
30+
def get_requirements(cls):
31+
# Since we're calling the plugin, make sure we have the plugin's requirements
32+
return [
33+
requirements.ModuleRequirement(
34+
name="kernel",
35+
description="Windows kernel",
36+
architectures=["Intel32", "Intel64"],
37+
),
38+
requirements.VersionRequirement(
39+
name="pslist", component=pslist.PsList, version=(2, 0, 0)
40+
),
41+
requirements.PluginRequirement(
42+
name="consoles", plugin=consoles.Consoles, version=(1, 0, 0)
43+
),
44+
requirements.BooleanRequirement(
45+
name="no_registry",
46+
description="Don't search the registry for possible values of CommandHistorySize",
47+
optional=True,
48+
default=False,
49+
),
50+
requirements.ListRequirement(
51+
name="max_history",
52+
element_type=int,
53+
description="CommandHistorySize values to search for.",
54+
optional=True,
55+
default=[50],
56+
),
57+
]
58+
59+
@classmethod
60+
def get_filtered_vads(
61+
cls,
62+
conhost_proc: interfaces.context.ContextInterface,
63+
size_filter: Optional[int] = 0x40000000,
64+
) -> Generator[Tuple[int, int], None, None]:
65+
"""
66+
Returns vads of a process with size smaller than size_filter
67+
68+
Args:
69+
conhost_proc: the process object for conhost.exe
70+
71+
Returns:
72+
A list of tuples of:
73+
vad_base: the base address
74+
vad_size: the size of the VAD
75+
"""
76+
for vad in conhost_proc.get_vad_root().traverse():
77+
base = vad.get_start()
78+
if vad.get_size() < size_filter:
79+
yield (base, vad.get_size())
80+
81+
@classmethod
82+
def get_command_history(
83+
cls,
84+
context: interfaces.context.ContextInterface,
85+
kernel_layer_name: str,
86+
kernel_symbol_table_name: str,
87+
config_path: str,
88+
procs: Generator[interfaces.objects.ObjectInterface, None, None],
89+
max_history: Set[int],
90+
) -> Tuple[
91+
interfaces.context.ContextInterface,
92+
interfaces.context.ContextInterface,
93+
Dict[str, Any],
94+
]:
95+
"""Gets the list of commands from each Command History structure
96+
97+
Args:
98+
context: The context to retrieve required elements (layers, symbol tables) from
99+
kernel_layer_name: The name of the layer on which to operate
100+
kernel_symbol_table_name: The name of the table containing the kernel symbols
101+
config_path: The config path where to find symbol files
102+
procs: list of process objects
103+
max_history: an initial set of CommandHistorySize values
104+
105+
Returns:
106+
The conhost process object, the command history structure, a dictionary of properties for
107+
that command history structure.
108+
"""
109+
110+
conhost_symbol_table = None
111+
112+
for conhost_proc, proc_layer_name in consoles.Consoles.find_conhost_proc(procs):
113+
if not conhost_proc:
114+
vollog.info(
115+
"Unable to find a valid conhost.exe process in the process list. Analysis cannot proceed."
116+
)
117+
continue
118+
vollog.debug(
119+
f"Found conhost process {conhost_proc} with pid {conhost_proc.UniqueProcessId}"
120+
)
121+
122+
conhostexe_base, conhostexe_size = consoles.Consoles.find_conhostexe(
123+
conhost_proc
124+
)
125+
if not conhostexe_base:
126+
vollog.info(
127+
"Unable to find the location of conhost.exe. Analysis cannot proceed."
128+
)
129+
continue
130+
vollog.debug(f"Found conhost.exe base at {conhostexe_base:#x}")
131+
132+
proc_layer = context.layers[proc_layer_name]
133+
134+
if conhost_symbol_table is None:
135+
conhost_symbol_table = consoles.Consoles.create_conhost_symbol_table(
136+
context,
137+
kernel_layer_name,
138+
kernel_symbol_table_name,
139+
config_path,
140+
proc_layer_name,
141+
conhostexe_base,
142+
)
143+
144+
conhost_module = context.module(
145+
conhost_symbol_table, proc_layer_name, offset=conhostexe_base
146+
)
147+
command_count_max_offset = conhost_module.get_type(
148+
"_COMMAND_HISTORY"
149+
).relative_child_offset("CommandCountMax")
150+
151+
sections = cls.get_filtered_vads(conhost_proc)
152+
found_history_for_proc = False
153+
# scan for potential _COMMAND_HISTORY structures by using the CommandHistorySize
154+
for max_history_value in max_history:
155+
max_history_bytes = struct.pack("H", max_history_value)
156+
vollog.debug(
157+
f"Scanning for CommandHistorySize value: {max_history_bytes}"
158+
)
159+
for address in proc_layer.scan(
160+
context,
161+
scanners.BytesScanner(max_history_bytes),
162+
sections=sections,
163+
):
164+
command_history = None
165+
command_history_properties = []
166+
167+
try:
168+
command_history = conhost_module.object(
169+
"_COMMAND_HISTORY",
170+
offset=address - command_count_max_offset,
171+
absolute=True,
172+
)
173+
174+
if not command_history.is_valid(max_history_value):
175+
continue
176+
177+
vollog.debug(
178+
f"Getting Command History properties for {command_history}"
179+
)
180+
command_history_properties.append(
181+
{
182+
"level": 0,
183+
"name": "_COMMAND_HISTORY",
184+
"address": command_history.vol.offset,
185+
"data": None,
186+
}
187+
)
188+
command_history_properties.append(
189+
{
190+
"level": 1,
191+
"name": "_COMMAND_HISTORY.Application",
192+
"address": command_history.Application.vol.offset,
193+
"data": command_history.get_application(),
194+
}
195+
)
196+
command_history_properties.append(
197+
{
198+
"level": 1,
199+
"name": "_COMMAND_HISTORY.ProcessHandle",
200+
"address": command_history.ConsoleProcessHandle.ProcessHandle.vol.offset,
201+
"data": hex(
202+
command_history.ConsoleProcessHandle.ProcessHandle
203+
),
204+
}
205+
)
206+
command_history_properties.append(
207+
{
208+
"level": 1,
209+
"name": "_COMMAND_HISTORY.CommandCount",
210+
"address": None,
211+
"data": command_history.CommandCount,
212+
}
213+
)
214+
command_history_properties.append(
215+
{
216+
"level": 1,
217+
"name": "_COMMAND_HISTORY.LastDisplayed",
218+
"address": command_history.LastDisplayed.vol.offset,
219+
"data": command_history.LastDisplayed,
220+
}
221+
)
222+
command_history_properties.append(
223+
{
224+
"level": 1,
225+
"name": "_COMMAND_HISTORY.CommandCountMax",
226+
"address": command_history.CommandCountMax.vol.offset,
227+
"data": command_history.CommandCountMax,
228+
}
229+
)
230+
231+
command_history_properties.append(
232+
{
233+
"level": 1,
234+
"name": "_COMMAND_HISTORY.CommandBucket",
235+
"address": command_history.CommandBucket.vol.offset,
236+
"data": "",
237+
}
238+
)
239+
for (
240+
cmd_index,
241+
bucket_cmd,
242+
) in command_history.scan_command_bucket():
243+
try:
244+
command_history_properties.append(
245+
{
246+
"level": 2,
247+
"name": f"_COMMAND_HISTORY.CommandBucket_Command_{cmd_index}",
248+
"address": bucket_cmd.vol.offset,
249+
"data": bucket_cmd.get_command_string(),
250+
}
251+
)
252+
except Exception as e:
253+
vollog.debug(
254+
f"reading {bucket_cmd} encountered exception {e}"
255+
)
256+
except Exception as e:
257+
vollog.debug(
258+
f"reading {command_history} encountered exception {e}"
259+
)
260+
261+
if command_history and command_history_properties:
262+
found_history_for_proc = True
263+
yield conhost_proc, command_history, command_history_properties
264+
265+
# if found_history_for_proc is still False, then none of the scanned locations found
266+
# a valid _COMMAND_HISTORY for the process, so yield the process and some empty data
267+
# so the process can at least be reported that it was found with no history
268+
if not found_history_for_proc:
269+
yield conhost_proc, command_history or None, []
270+
271+
def _generator(
272+
self, procs: Generator[interfaces.objects.ObjectInterface, None, None]
273+
):
274+
"""
275+
Generates the command history to use in rendering
276+
277+
Args:
278+
procs: the process list filtered to conhost.exe instances
279+
"""
280+
281+
kernel = self.context.modules[self.config["kernel"]]
282+
283+
max_history = set(self.config.get("max_history", [50]))
284+
no_registry = self.config.get("no_registry")
285+
286+
if no_registry is False:
287+
max_history, _ = consoles.Consoles.get_console_settings_from_registry(
288+
self.context,
289+
self.config_path,
290+
kernel.layer_name,
291+
kernel.symbol_table_name,
292+
max_history,
293+
[],
294+
)
295+
296+
vollog.debug(f"Possible CommandHistorySize values: {max_history}")
297+
298+
proc = None
299+
for (
300+
proc,
301+
command_history,
302+
command_history_properties,
303+
) in self.get_command_history(
304+
self.context,
305+
kernel.layer_name,
306+
kernel.symbol_table_name,
307+
self.config_path,
308+
procs,
309+
max_history,
310+
):
311+
process_name = utility.array_to_string(proc.ImageFileName)
312+
process_pid = proc.UniqueProcessId
313+
314+
if command_history and command_history_properties:
315+
for command_history_property in command_history_properties:
316+
yield (
317+
command_history_property["level"],
318+
(
319+
process_pid,
320+
process_name,
321+
format_hints.Hex(command_history.vol.offset),
322+
command_history_property["name"],
323+
(
324+
renderers.NotApplicableValue()
325+
if command_history_property["address"] is None
326+
else format_hints.Hex(
327+
command_history_property["address"]
328+
)
329+
),
330+
str(command_history_property["data"]),
331+
),
332+
)
333+
else:
334+
yield (
335+
0,
336+
(
337+
process_pid,
338+
process_name,
339+
(
340+
format_hints.Hex(command_history.vol.offset)
341+
if command_history
342+
else renderers.NotApplicableValue()
343+
),
344+
"_COMMAND_HISTORY",
345+
renderers.NotApplicableValue(),
346+
"History Not Found",
347+
),
348+
)
349+
350+
if proc is None:
351+
vollog.warn("No conhost.exe processes found.")
352+
353+
def _conhost_proc_filter(self, proc: interfaces.objects.ObjectInterface):
354+
"""
355+
Used to filter to only conhost.exe processes
356+
"""
357+
process_name = utility.array_to_string(proc.ImageFileName)
358+
359+
return process_name != "conhost.exe"
360+
361+
def run(self):
362+
kernel = self.context.modules[self.config["kernel"]]
363+
364+
return renderers.TreeGrid(
365+
[
366+
("PID", int),
367+
("Process", str),
368+
("ConsoleInfo", format_hints.Hex),
369+
("Property", str),
370+
("Address", format_hints.Hex),
371+
("Data", str),
372+
],
373+
self._generator(
374+
pslist.PsList.list_processes(
375+
context=self.context,
376+
layer_name=kernel.layer_name,
377+
symbol_table=kernel.symbol_table_name,
378+
filter_func=self._conhost_proc_filter,
379+
)
380+
),
381+
)

0 commit comments

Comments
 (0)