mirror of
https://github.com/Kowntaewook/MCPTrace.git
synced 2026-11-02 04:45:09 +01:00
kwonta3wook
031b8ab298
Add a Linux task/VMA adapter and POSIX path policy checks. Share scanning and output with Windows while preserving its existing VAD fields. Report skipped address spaces and scan coverage, and test both adapters across host platforms.
222 lines
8.9 KiB
Python
222 lines
8.9 KiB
Python
"""Linux adapter integration with real Volatility layers and synthetic tasks/VMAs."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from test_core import encode, request
|
|
from test_plugin import ROOT, GappedLayer, plugin_module, rows_of
|
|
from volatility3.cli import CommandLine
|
|
from volatility3.framework import contexts, exceptions
|
|
from volatility3.framework.layers.physical import BufferDataLayer
|
|
|
|
|
|
def make_task(context, pid, data, ranges=None, layer_type=BufferDataLayer):
|
|
layer_name = f"task-{pid}"
|
|
context.layers.add_layer(layer_type(context, layer_name, layer_name, data))
|
|
ranges = ranges if ranges is not None else [(0, len(data))]
|
|
vmas = [SimpleNamespace(vm_start=start, vm_end=end) for start, end in ranges]
|
|
return SimpleNamespace(
|
|
pid=pid,
|
|
tgid=pid,
|
|
comm="mcp-server",
|
|
mm=SimpleNamespace(get_vma_iter=lambda: iter(vmas)),
|
|
add_process_layer=lambda: layer_name,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def linux_adapter(monkeypatch):
|
|
module = plugin_module("linux")
|
|
context = contexts.Context()
|
|
req = encode(request(tool="read_file", arguments={"path": "/etc/shadow"}))
|
|
resp = encode({"jsonrpc": "2.0", "id": 1, "result": {"content": []}}, "utf-16-le")
|
|
memory = b"\0" * 256 + req + b"\0" * 101 + resp + b"\0" * 256
|
|
tasks = [make_task(context, 42, memory)]
|
|
|
|
def list_tasks(*, context, vmlinux_module_name, filter_func, include_threads):
|
|
assert vmlinux_module_name == "synthetic-kernel"
|
|
assert include_threads is False
|
|
return (task for task in tasks if not filter_func(task))
|
|
|
|
monkeypatch.setattr(module.pslist.PsList, "list_tasks", list_tasks)
|
|
monkeypatch.setattr(module.utility, "array_to_string", lambda value: value)
|
|
requirements = module.MCPTrace.get_requirements()
|
|
# Kernel objects are test doubles; layer reading, filtering, parsing, TreeGrid,
|
|
# and the JSONL file handler use the real installed Volatility implementation.
|
|
monkeypatch.setattr(module.MCPTrace, "get_requirements", classmethod(lambda cls: []))
|
|
config = "plugins.LinuxMCPTrace"
|
|
context.config[f"{config}.kernel"] = "synthetic-kernel"
|
|
for requirement in requirements:
|
|
if requirement.name not in {"kernel", "pslist"}:
|
|
context.config[f"{config}.{requirement.name}"] = requirement.default
|
|
return module.MCPTrace(context, config), tasks
|
|
|
|
|
|
def summary(rows):
|
|
return json.loads(rows[-1][13])
|
|
|
|
|
|
def test_linux_extracts_both_encodings_and_posix_policy(linux_adapter):
|
|
plugin, _ = linux_adapter
|
|
plugin.config["policy"] = str(ROOT / "examples" / "policy-linux.json")
|
|
grid = plugin.run()
|
|
assert "VMAStart" in [column.name for column in grid.columns]
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == 3
|
|
assert {row[4] for row in rows[:-1]} == {"utf-8", "utf-16-le"}
|
|
assert all(row[0] == 42 and row[10] == "candidate-only" for row in rows[:-1])
|
|
assert rows[0][3] == 256
|
|
assert json.loads(rows[0][12])[0]["status"] == "policy-match"
|
|
assert summary(rows)["target_os"] == "linux"
|
|
assert summary(rows)["regions_scanned"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize("selected,expected", [([42], {42}), ([99], {99}), ([123], set())])
|
|
def test_linux_pid_filter(linux_adapter, selected, expected):
|
|
plugin, tasks = linux_adapter
|
|
tasks.append(make_task(plugin.context, 99, encode(request(99))))
|
|
plugin.config["pid"] = selected
|
|
assert {row[0] for row in rows_of(plugin)} == expected
|
|
|
|
|
|
def test_linux_does_not_link_ids_between_processes(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
tasks[:] = [
|
|
make_task(plugin.context, 100, encode(request(8))),
|
|
make_task(plugin.context, 101, encode({"jsonrpc": "2.0", "id": 8, "result": {}})),
|
|
]
|
|
plugin.config["include-rpc"] = True
|
|
rows = [row for row in rows_of(plugin) if row[5] == "complete"]
|
|
assert len(rows) == 2 and all(row[10] == "unpaired" for row in rows)
|
|
|
|
|
|
def test_linux_kernel_thread_never_uses_borrowed_mm(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
tasks[0].active_mm = tasks[0].mm
|
|
tasks[0].mm = None
|
|
|
|
def forbidden_layer():
|
|
pytest.fail("No process layer should be constructed for a kernel thread")
|
|
|
|
tasks[0].add_process_layer = forbidden_layer
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == 1
|
|
assert summary(rows)["skipped_reason"] == "no-user-address-space"
|
|
assert summary(rows)["bytes_read"] == 0 and summary(rows)["error"] is None
|
|
|
|
|
|
def test_linux_failed_process_layer_does_not_abort_next_process(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
tasks[0].add_process_layer = lambda: None
|
|
tasks.append(make_task(plugin.context, 99, encode(request(99))))
|
|
rows = rows_of(plugin)
|
|
assert json.loads(rows[0][13])["skipped_reason"] == "no-process-layer"
|
|
assert any(row[0] == 99 and row[5] == "complete" for row in rows)
|
|
|
|
|
|
def test_linux_vm_end_is_exclusive(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
first, second = encode(request(1)), encode(request(2))
|
|
start = 123
|
|
tasks[:] = [
|
|
make_task(plugin.context, 99, b"\0" * start + first + second, [(start, start + len(first))])
|
|
]
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == 2 and json.loads(rows[0][7]) == 1
|
|
assert rows[0][2] == start and summary(rows)["bytes_read"] == len(first)
|
|
|
|
|
|
def test_linux_adjacent_vmas_are_not_joined(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
raw = encode(request(text="x" * 200))
|
|
tasks[:] = [make_task(plugin.context, 99, raw, [(0, 80), (80, len(raw))])]
|
|
plugin.config["include-partial"] = True
|
|
rows = rows_of(plugin)
|
|
assert any(row[5] == "fragment" for row in rows)
|
|
assert all(row[5] != "complete" for row in rows)
|
|
assert summary(rows)["regions_scanned"] == 2
|
|
|
|
|
|
def test_linux_missing_page_does_not_bridge_messages(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
raw = encode(request(1, text="x" * 200))
|
|
intact = encode(request(2))
|
|
data = bytearray(12288)
|
|
data[4000:4096] = raw[:96]
|
|
data[8192 : 8192 + len(raw[96:])] = raw[96:]
|
|
data[9000 : 9000 + len(intact)] = intact
|
|
tasks[:] = [make_task(plugin.context, 99, bytes(data), layer_type=GappedLayer)]
|
|
plugin.config["include-partial"] = True
|
|
rows = rows_of(plugin)
|
|
assert [json.loads(row[7]) for row in rows if row[5] == "complete"] == [2]
|
|
assert summary(rows)["unreadable_bytes"] == 4096
|
|
assert summary(rows)["bytes_read"] == 8192
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["unreadable", "unsupported-layout"])
|
|
def test_linux_later_vma_failure_preserves_existing_evidence(linux_adapter, failure):
|
|
plugin, tasks = linux_adapter
|
|
first = next(tasks[0].mm.get_vma_iter())
|
|
|
|
def broken_vmas():
|
|
yield first
|
|
if failure == "unreadable":
|
|
raise exceptions.InvalidAddressException("synthetic-kernel", 0xBAD)
|
|
raise AttributeError("Unable to find mmap or mm_mt in mm_struct")
|
|
|
|
tasks[0].mm.get_vma_iter = broken_vmas
|
|
rows = rows_of(plugin)
|
|
assert len([row for row in rows if row[5] == "complete"]) == 2
|
|
assert summary(rows)["limits_hit"] == ["process-read-error"]
|
|
assert summary(rows)["error"] is not None
|
|
|
|
|
|
def test_linux_invalid_vma_ranges_are_reported(linux_adapter):
|
|
plugin, tasks = linux_adapter
|
|
raw = encode(request())
|
|
tasks[:] = [make_task(plugin.context, 99, raw, [(-1, 1), (0, 0), (2, 1), (0, len(raw))])]
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == 2 and summary(rows)["invalid_regions"] == 3
|
|
|
|
|
|
@pytest.mark.parametrize("option,limit,count", [("max-bytes", 128, 0), ("max-records", 1, 1)])
|
|
def test_linux_limits_are_visible(linux_adapter, option, limit, count):
|
|
plugin, _ = linux_adapter
|
|
plugin.config[option] = limit
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == count + 1
|
|
assert summary(rows)["limits_hit"] == [option]
|
|
|
|
|
|
def test_linux_suspicious_only_with_posix_policy(linux_adapter):
|
|
plugin, _ = linux_adapter
|
|
plugin.config["policy"] = str(ROOT / "examples" / "policy-linux.json")
|
|
plugin.config["suspicious-only"] = True
|
|
rows = rows_of(plugin)
|
|
assert len(rows) == 2 and rows[0][6] == "request"
|
|
|
|
|
|
def test_linux_jsonl_records_vma_provenance_and_exact_raw(linux_adapter, tmp_path):
|
|
plugin, _ = linux_adapter
|
|
cli = CommandLine()
|
|
cli.output_dir = str(tmp_path)
|
|
plugin.set_open_method(cli.file_handler_class_factory())
|
|
plugin.config["dump"] = True
|
|
rows_of(plugin)
|
|
exported = [
|
|
json.loads(line)
|
|
for line in (tmp_path / "mcptrace.pid-42.jsonl").read_text(encoding="utf-8").splitlines()
|
|
]
|
|
assert exported[0]["target_os"] == "linux" and exported[0]["region_kind"] == "vma"
|
|
records = exported[1:-1]
|
|
assert len(records) == 2 and exported[-1]["type"] == "scan-summary"
|
|
for item in records:
|
|
assert item["target_os"] == "linux" and item["region_kind"] == "vma"
|
|
assert item["vma_start"] == item["region_start"] == 0 and "vad_start" not in item
|
|
raw = base64.b64decode(item["raw_base64"])
|
|
assert hashlib.sha256(raw).hexdigest() == item["raw_sha256"]
|
|
assert json.loads(raw.decode(item["encoding"])) == item["message"]
|