mirror of
https://github.com/Kowntaewook/MCPTrace.git
synced 2026-11-02 04:45:09 +01:00
kwonta3wook
fbcb4a92ac
Carve MCP/JSON-RPC candidates with original bytes and provenance, conservative ID correlation, opt-in argument policies, JSONL export, and scan coverage. Add synthetic Volatility integration tests, CI, and Korean documentation.
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import copy
|
|
|
|
import pytest
|
|
from test_core import carve, request
|
|
|
|
from mcptrace.policy import evaluate, validate_policy
|
|
|
|
|
|
def rule(operator, value, argument="/path"):
|
|
return {
|
|
"id": "R1",
|
|
"description": "Analyst-defined policy",
|
|
"severity": "medium",
|
|
"tool": "read_file",
|
|
"argument": argument,
|
|
"operator": operator,
|
|
"value": value,
|
|
}
|
|
|
|
|
|
def run_rule(value, operator="windows_path_outside", expected=None):
|
|
spec = rule(operator, expected if expected is not None else [r"C:\Cases"])
|
|
rules = validate_policy({"version": 1, "rules": [spec]})
|
|
return evaluate(carve(request(tool="read_file", arguments={"path": value})), rules)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
[r"C:\Cases\evidence.txt", r"c:\CASES\sub\a.txt", r"C:/Cases/a.txt", r"\\?\C:\Cases\a.txt"],
|
|
)
|
|
def test_allowed_path_boundaries(value):
|
|
assert run_rule(value) == []
|
|
|
|
|
|
@pytest.mark.parametrize("value", [r"C:\Cases2\a.txt", r"C:\Cases\..\secret.txt", r"D:\Cases\a"])
|
|
def test_outside_paths_match(value):
|
|
assert run_rule(value)[0]["status"] == "policy-match"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
[
|
|
r"C:relative.txt",
|
|
r"\rooted.txt",
|
|
r"%USERPROFILE%\a",
|
|
3,
|
|
None,
|
|
r"C:\Cases\a.txt:secret",
|
|
r"C:\Cases\a. ",
|
|
r"C:\Cases\*.txt",
|
|
],
|
|
)
|
|
def test_unresolved_inputs_are_not_marked_safe(value):
|
|
assert run_rule(value)[0]["status"] == "not-evaluated"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,matched",
|
|
[
|
|
("https://API.EXAMPLE.TEST./x", False),
|
|
("https://api.example.test.evil.test/x", True),
|
|
("https://api.example.test@evil.test/x", True),
|
|
("https://evil.test/?allowed=api.example.test", True),
|
|
("https://api.example.test:443/path", False),
|
|
],
|
|
)
|
|
def test_url_host_uses_parsed_authority(url, matched):
|
|
findings = run_rule(url, "url_host_not_in", ["api.example.test"])
|
|
assert bool(findings) == matched
|
|
|
|
|
|
def test_policy_requires_exact_tool_and_complete_request():
|
|
rules = validate_policy({"version": 1, "rules": [rule("contains", "secret")]})
|
|
assert evaluate(carve(request(tool="another_tool", arguments={"path": "secret"})), rules) == []
|
|
record = carve(request(tool="read_file", arguments={"path": "secret"}))
|
|
assert evaluate(record, rules)[0]["status"] == "policy-match"
|
|
record.state = "fragment"
|
|
assert evaluate(record, rules) == []
|
|
|
|
|
|
def test_json_pointer_and_missing_arguments():
|
|
spec = rule("equals", "secret", "/nested/a~1b/0")
|
|
rules = validate_policy({"version": 1, "rules": [spec]})
|
|
rec = carve(request(tool="read_file", arguments={"nested": {"a/b": ["secret"]}}))
|
|
assert evaluate(rec, rules)[0]["status"] == "policy-match"
|
|
rec.obj["params"]["arguments"] = {}
|
|
assert evaluate(rec, rules)[0]["status"] == "not-evaluated"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
[
|
|
{"operator": "execute"},
|
|
{"severity": "critical"},
|
|
{"extra": "ignored?"},
|
|
{"argument": "path"},
|
|
{"value": []},
|
|
{"value": ["relative"]},
|
|
{"argument": "/bad~2pointer"},
|
|
],
|
|
)
|
|
def test_bad_policy_is_rejected(change):
|
|
spec = rule("windows_path_outside", [r"C:\Cases"])
|
|
spec.update(change)
|
|
with pytest.raises(ValueError):
|
|
validate_policy({"version": 1, "rules": [spec]})
|
|
|
|
|
|
def test_duplicate_rule_ids_rejected():
|
|
spec = rule("contains", "x")
|
|
with pytest.raises(ValueError):
|
|
validate_policy({"version": 1, "rules": [spec, copy.deepcopy(spec)]})
|