chenmingxuan/Artemis-Plugins
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

fix(device-test): contain tool processes and add reviewed preview

Browse Source
This commit is contained in:
chenmingxuan committed 2026-10-30 12:43:36 +08:00
1 parent c98d7527a0
commit 7d31e09cb7
59 files changed
+23722
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

plugins/artemis-device-test/skill/tests/test_tool_runner.py
+417
View File
@@ -0,0 +1,417 @@
"""Offline safety/command tests. No test contacts a device or starts a tool."""
import contextlib
import importlib.util
import io
import json
import pathlib
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "tool_runner.py"
runner = None
if SCRIPT.exists():
spec = importlib.util.spec_from_file_location("artemis_tool_runner", SCRIPT)
runner = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = runner
spec.loader.exec_module(runner)
class ToolRunnerTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(runner, "tool_runner implementation is required")
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = pathlib.Path(self.tmp.name)
self.tools = {}
for name in ("ffprobe", "ffmpeg", "tshark", "dumpcap", "nmap", "binwalk", "powershell"):
path = self.root / (name + ".exe")
path.touch()
self.tools[name] = str(path)
self.profile = {"tools": self.tools}
self.media = self.root / "input media.bin"
self.media.write_bytes(b"offline fixture")
def plan(self, *args, env=None, profile=None):
return runner.build_plan(runner.parse_args(list(args)), self.profile if profile is None else profile, env or {})
def test_scan_constructs_one_host_explicit_ports_no_scripts(self):
p = self.plan("scan", "--host", "192.0.2.10", "--ports", "554,80,22")
self.assertEqual(p.tool_id, "nmap")
self.assertIn("-sT", p.argv)
self.assertIn("-n", p.argv)
self.assertEqual(p.argv[-1], "192.0.2.10")
self.assertEqual(p.argv[p.argv.index("-p") + 1], "22,80,554")
self.assertNotIn("-sV", p.argv)
self.assertNotIn("-A", p.argv)
self.assertFalse(any("script" in x for x in p.argv))
def test_nmap_preserves_aggregate_states_without_assigning_ports(self):
p = self.plan("scan", "--host", "192.0.2.10", "--ports", "22,80,554")
xml = '''<nmaprun><host><status state="up"/><ports>
<extraports state="closed" count="1"><extrareasons reason="conn-refused" count="1"/></extraports>
<extraports state="filtered" count="1"><extrareasons reason="no-response" count="1"/></extraports>
<port protocol="tcp" portid="22"><state state="open"/></port>
</ports></host></nmaprun>'''
data = runner.safe_data(p, xml)
self.assertEqual(data.get("requested_ports"), [22, 80, 554])
host = data["hosts"][0]
self.assertEqual(host["aggregate_states"], [
{"state": "closed", "count": 1, "reasons": [{"reason": "conn-refused", "count": 1}]},
{"state": "filtered", "count": 1, "reasons": [{"reason": "no-response", "count": 1}]},
])
self.assertEqual(host["ports_not_individually_reported"], [80, 554])
self.assertEqual(host["individual_port_coverage"], "partial")
self.assertEqual(host["ports"], [{"port": "22", "protocol": "tcp", "state": "open"}])
def test_nmap_aggregate_only_is_not_individual_port_coverage(self):
p = self.plan("scan", "--host", "192.0.2.10", "--ports", "80,554")
data = runner.safe_data(p, '<nmaprun><host><status state="up"/><ports><extraports state="closed" count="2"/></ports></host></nmaprun>')
host = data["hosts"][0]
self.assertEqual(host.get("aggregate_states"), [{"state": "closed", "count": 2, "reasons": []}])
self.assertEqual(host["ports"], [])
self.assertEqual(host["ports_not_individually_reported"], [80, 554])
self.assertEqual(host["individual_port_coverage"], "none")
def test_nmap_missing_results_remain_explicit(self):
p = self.plan("scan", "--host", "192.0.2.10", "--ports", "22,80")
data = runner.safe_data(p, '<nmaprun><host><status state="up"/><ports><port protocol="tcp" portid="22"><state state="open"/></port></ports></host></nmaprun>')
self.assertEqual(data["hosts"][0].get("aggregate_states"), [])
self.assertEqual(data["hosts"][0]["ports_not_individually_reported"], [80])
absent = runner.safe_data(p, '<nmaprun/>')
self.assertEqual(absent["requested_ports"], [22, 80])
self.assertEqual(absent["hosts"], [])
self.assertFalse(absent["host_results_present"])
def test_scan_rejects_hostnames_networks_and_option_injection(self):
for value in ("192.0.2.0/24", "camera.local", "192.0.2.1 192.0.2.2", "-iLtargets", "224.0.0.1", "0.0.0.0"):
with self.subTest(value=value), self.assertRaises(runner.ValidationError):
self.plan("scan", "--host", value, "--ports", "80")
def test_scan_rejects_ranges_unbounded_and_injected_ports(self):
for value in ("1-65535", "0", "65536", "80;echo", "80 --script vuln", "", "T:80"):
with self.subTest(value=value), self.assertRaises(runner.ValidationError):
self.plan("scan", "--host", "192.0.2.1", "--ports", value)
def test_no_arbitrary_args_or_direct_rtsp_url_cli(self):
for args in (("scan", "--host", "192.0.2.1", "--ports", "80", "--args", "-A"),
("probe", "--url", "rtsp://user:secret@192.0.2.1/live")):
with self.subTest(args=args), self.assertRaises(runner.ValidationError):
runner.parse_args(list(args))
def test_capture_limits_interface_host_time_and_size(self):
p = self.plan("capture", "--interface", "1", "--host", "192.0.2.1", "--duration", "5", "--output", str(self.root / "capture.pcapng"))
self.assertEqual(p.argv[p.argv.index("-f") + 1], "host 192.0.2.1")
self.assertIn("duration:5", p.argv)
self.assertIn("filesize:102400", p.argv)
self.assertIn("-s", p.argv)
self.assertGreater(p.timeout, 5)
def test_capture_rejects_invalid_duration_interface_and_overwrite(self):
common = ["capture", "--interface", "1", "--host", "192.0.2.1", "--duration", "5", "--output", str(self.root / "out.pcapng")]
for index, value in ((2, "-D"), (2, "rpcap://host"), (6, "0"), (6, "601"), (8, str(self.media))):
args = common.copy()
args[index] = value
with self.subTest(value=value), self.assertRaises(runner.ValidationError):
self.plan(*args)
def test_tshark_reads_offline_only_with_safe_fields(self):
p = self.plan("analyze-pcap", "--input", str(self.media), "--max-packets", "10")
self.assertIn("-r", p.argv)
self.assertNotIn("-i", p.argv)
self.assertIn("fields", p.argv)
self.assertIn("ip.src", p.argv)
self.assertNotIn("http.authorization", p.argv)
self.assertNotIn("tcp.payload", p.argv)
def test_probe_local_input_is_absolute_and_network_protocols_disabled(self):
p = self.plan("probe", "--input", str(self.media))
self.assertEqual(p.argv[-1], str(self.media.resolve()))
self.assertIn("-protocol_whitelist", p.argv)
self.assertEqual(p.argv[p.argv.index("-protocol_whitelist") + 1], "file")
self.assertNotIn("-show_format", p.argv)
def test_rtsp_requires_env_and_redacts_all_url_in_plan(self):
url = "rtsp://admin:TopSecret123@192.0.2.1:554/live?token=HiddenToken"
p = self.plan("probe", "--rtsp-env", "CAMERA_RTSP", env={"CAMERA_RTSP": url})
self.assertEqual(p.argv[-1], url)
public = json.dumps(p.public()) + repr(p)
self.assertNotIn("TopSecret123", public)
self.assertNotIn("HiddenToken", public)
self.assertNotIn(url, public)
self.assertIn("CAMERA_RTSP", public)
def test_rtsp_rejects_absent_env_non_rtsp_and_bad_url(self):
for env in ({}, {"URL": "http://192.0.2.1"}, {"URL": "rtsp://"}, {"URL": "rtsp://192.0.2.1:99999/live"}, {"URL": "rtsp://192.0.2.1/live\n-x"}):
with self.subTest(env=env), self.assertRaises(runner.ValidationError):
self.plan("probe", "--rtsp-env", "URL", env=env)
def test_profile_rejects_plaintext_secrets_and_auth_urls(self):
for data in ({"password": "secret"}, {"device": {"password": "secret"}}, {"rtsp_url": "rtsp://a:b@192.0.2.1/live"}, {"tools": {"nmap": {"args": "-A"}}}):
with self.subTest(data=data), self.assertRaises(runner.ValidationError):
runner.validate_profile(data)
runner.validate_profile({"auth": {"password_env": "DEVICE_PASSWORD"}, "tools": {"nmap": self.tools["nmap"]}})
def test_profile_rejects_compound_secret_keys_and_token_urls(self):
for data in ({"device_password": "secret"}, {"authToken": "secret"}, {"url": "https://example.invalid/?access_token=secret"}, {"密码": "secret"}):
with self.subTest(data=data), self.assertRaises(runner.ValidationError):
runner.validate_profile(data)
def test_probe_sanitization_handles_quotes_in_secret_without_echo(self):
url = "rtsp://admin:Se%22cret@192.0.2.1/live"
p = self.plan("probe", "--rtsp-env", "URL", env={"URL": url})
data = runner.safe_data(p, json.dumps({"streams": [{"codec_name": 'Se"cret'}]}))
self.assertNotIn('Se"cret', json.dumps(data).replace('\\"', '"'))
def test_missing_tool_never_falls_back_to_installers(self):
with mock.patch.object(runner.shutil, "which", return_value=None):
with self.assertRaises(runner.ToolMissing):
self.plan("scan", "--host", "192.0.2.1", "--ports", "80", profile={})
installer = self.root / "nmap-setup.exe"
installer.touch()
with self.assertRaises(runner.ValidationError):
self.plan("scan", "--host", "192.0.2.1", "--ports", "80", profile={"tools": {"nmap": str(installer)}})
def test_tool_root_finds_extracted_binary_but_not_zip_members(self):
args = runner.parse_args(["--tool-root", str(self.root), "scan", "--host", "192.0.2.1", "--ports", "80"])
with mock.patch.object(runner.shutil, "which", return_value=None):
self.assertEqual(runner.build_plan(args, {}, {}).argv[0], self.tools["nmap"])
def test_firmware_scan_never_extracts_or_executes(self):
p = self.plan("firmware-scan", "--input", str(self.media))
self.assertEqual(list(p.argv), [self.tools["binwalk"], str(self.media.resolve())])
def test_binwalk_runtime_uses_only_fixed_isolated_module_invocation(self):
python = self.root / "python.exe"
python.touch()
p = self.plan("firmware-scan", "--input", str(self.media),
profile={"runtimes": {"binwalk_python": str(python)}})
self.assertEqual(p.argv, [str(python), "-I", "-m", "binwalk", str(self.media)])
def test_runtime_and_library_configuration_rejects_arbitrary_launchers(self):
for profile in ({"runtimes": {"binwalk_python": {"args": "-c evil"}}},
{"runtimes": {"other_python": "python.exe"}},
{"libraries": {"mediainfo": {"args": "evil"}}},
{"libraries": {"other": "unknown.dll"}}):
with self.subTest(profile=profile), self.assertRaises(runner.ValidationError):
runner.validate_profile(profile)
cmd = self.root / "cmd.exe"
cmd.touch()
with self.assertRaises(runner.ValidationError):
self.plan("firmware-scan", "--input", str(self.media),
profile={"runtimes": {"binwalk_python": str(cmd)}})
def test_profile_resolves_library_and_runtime_relative_to_profile(self):
profile = self.root / "profile.json"
profile.write_text(json.dumps({"runtimes": {"binwalk_python": "env/Scripts/python.exe"},
"libraries": {"mediainfo": "media/MediaInfo.dll"}}))
loaded = runner.load_profile(str(profile))
self.assertEqual(loaded["runtimes"]["binwalk_python"], str(self.root / "env/Scripts/python.exe"))
self.assertEqual(loaded["libraries"]["mediainfo"], str(self.root / "media/MediaInfo.dll"))
def test_binwalk_scan_does_not_load_user_config_or_pythonpath(self):
p = self.plan("firmware-scan", "--input", str(self.media))
seen = {}
def run(argv, **kwargs):
seen.update(kwargs)
config = pathlib.Path(kwargs["env"]["XDG_CONFIG_HOME"])
self.assertTrue(config.is_dir())
self.assertEqual(list(config.iterdir()), [])
return subprocess.CompletedProcess(argv, 0, "signature", "")
with mock.patch.dict(runner.os.environ, {"XDG_CONFIG_HOME": "existing-config", "PYTHONPATH": "unsafe"}), \
mock.patch.object(runner, "_run_in_windows_job", side_effect=run):
result = runner.execute(p)
self.assertEqual(result["execution_status"], "completed")
self.assertNotIn("PYTHONPATH", seen["env"])
self.assertFalse(pathlib.Path(seen["env"]["XDG_CONFIG_HOME"]).exists())
def test_media_info_is_bounded_child_with_local_dll_and_file_only(self):
library = self.root / "MediaInfo.dll"
library.touch()
profile = {"libraries": {"mediainfo": str(library)}}
p = self.plan("media-info", "--input", str(self.media), profile=profile)
self.assertEqual(p.tool_id, "mediainfo")
self.assertEqual(p.argv[:4], [sys.executable, "-I", str(SCRIPT), "_media-info-worker"])
self.assertEqual(p.argv[-4:], ["--library", str(library), "--input", str(self.media)])
self.assertGreater(p.timeout, 0)
for value in ("rtsp://192.0.2.1/live", "https://example.invalid/video.mp4", "//server/media.mp4"):
with self.subTest(value=value), self.assertRaises(runner.ValidationError):
self.plan("media-info", "--input", value, profile=profile)
def test_media_info_filters_fields_including_embedded_title_and_filename(self):
p = runner.Plan("media-info", "mediainfo", [], 10)
raw = {"version": "MediaInfoLib - v21.09", "general": {"Format": "MPEG-4", "CompleteName": "private-file"},
"video": [{"Width": "160", "Height": "120", "FrameCount": "5", "Title": "private-title"}],
"audio": [{"SamplingRate": "8000", "Channels": "1"}], "other": "secret"}
data = runner.safe_data(p, json.dumps(raw))
self.assertEqual(data["video"][0]["Width"], "160")
self.assertEqual(data["audio"][0]["SamplingRate"], "8000")
self.assertNotIn("private", json.dumps(data))
self.assertNotIn("other", data)
def test_media_info_missing_dll_does_not_fall_back_to_gui_installer(self):
with self.assertRaises(runner.ToolMissing):
self.plan("media-info", "--input", str(self.media), profile={})
installer = self.root / "MediaInfo_GUI.exe"
installer.touch()
with self.assertRaises(runner.ValidationError):
self.plan("media-info", "--input", str(self.media),
profile={"libraries": {"mediainfo": str(installer)}})
def test_media_info_worker_disables_internet_using_empty_value_and_checks_it(self):
library = self.root / "MediaInfo.dll"
library.touch()
lib = mock.MagicMock()
states = {None: "1", 123: "1"}
def option(handle, name, value):
if name == "Internet":
# MediaInfo 21.09 treats every nonempty value, including "No",
# as enabled. An empty value is the documented implementation.
states[handle] = "1" if value else ""
return ""
if name == "Internet_Get":
return states[handle]
return "MediaInfoLib - v21.09"
lib.MediaInfo_Option.side_effect = option
lib.MediaInfo_New.return_value = 123
lib.MediaInfo_Open.return_value = 1
lib.MediaInfo_Count_Get.return_value = 0
output = io.StringIO()
with mock.patch("ctypes.WinDLL", return_value=lib, create=True), contextlib.redirect_stdout(output):
code = runner.media_info_worker(["--library", str(library), "--input", str(self.media)])
self.assertEqual(code, 0)
self.assertEqual(states, {None: "", 123: ""})
for handle in (None, 123):
self.assertIn(mock.call(handle, "Internet_Get", ""), lib.MediaInfo_Option.call_args_list)
lib.MediaInfo_Open.assert_called_once_with(123, str(self.media))
def test_media_info_worker_refuses_input_when_internet_disabled_state_is_unconfirmed(self):
library = self.root / "MediaInfo.dll"
library.touch()
for blocked_handle in (None, 123):
for unexpected_state in ("1", "Option not known", None):
with self.subTest(handle=blocked_handle, state=unexpected_state):
lib = mock.MagicMock()
def option(handle, name, value):
if name == "Internet_Get":
return unexpected_state if handle == blocked_handle else ""
return ""
lib.MediaInfo_Option.side_effect = option
lib.MediaInfo_New.return_value = 123
lib.MediaInfo_Open.return_value = 1
lib.MediaInfo_Count_Get.return_value = 0
with mock.patch("ctypes.WinDLL", return_value=lib, create=True), contextlib.redirect_stdout(io.StringIO()):
code = runner.media_info_worker(["--library", str(library), "--input", str(self.media)])
self.assertEqual(code, 1)
lib.MediaInfo_Open.assert_not_called()
def test_short_rtsp_query_secret_does_not_corrupt_media_measurements(self):
url = "rtsp://admin:LongPassword@192.0.2.1/live?token=1&channel=0"
p = self.plan("probe", "--rtsp-env", "URL", env={"URL": url})
raw = {"streams": [{"codec_name": "hevc", "avg_frame_rate": "20/1", "sample_rate": "8000"}],
"format": {"duration": "1.0", "bit_rate": "123456"}}
self.assertEqual(runner.safe_data(p, json.dumps(raw)), raw)
self.assertNotIn(url, json.dumps(p.public()))
def test_serial_is_bounded_read_only_and_no_handshake_flags(self):
p = self.plan("serial", "--port", "COM9", "--baud", "115200", "--duration", "3", "--output", str(self.root / "serial.log"))
self.assertIn("-NoProfile", p.argv)
self.assertIn("-DurationSeconds", p.argv)
self.assertIn("3", p.argv)
self.assertNotIn("-DtrEnable", p.argv)
self.assertNotIn("-RtsEnable", p.argv)
self.assertNotIn("-Command", p.argv)
self.assertTrue(any(x.endswith("serial_monitor.ps1") for x in p.argv))
def test_serial_control_lines_require_explicit_independent_flags(self):
for flags, expected in ((["--dtr-enable"], {"-DtrEnable"}),
(["--rts-enable"], {"-RtsEnable"}),
(["--dtr-enable", "--rts-enable"], {"-DtrEnable", "-RtsEnable"})):
with self.subTest(flags=flags):
p = self.plan("serial", "--port", "COM9", "--baud", "115200", "--duration", "3",
"--output", str(self.root / "serial.log"), *flags)
self.assertEqual(set(p.argv) & {"-DtrEnable", "-RtsEnable"}, expected)
self.assertEqual(p.metadata["dtr_enabled"], "-DtrEnable" in expected)
self.assertEqual(p.metadata["rts_enabled"], "-RtsEnable" in expected)
self.assertNotIn("-Handshake", p.argv)
self.assertNotIn("-Command", p.argv)
def test_timeout_must_be_finite_positive_and_cover_capture(self):
for value in ("0", "-1", "3601", "nan", "inf"):
with self.subTest(value=value), self.assertRaises(runner.ValidationError):
self.plan("--timeout", value, "firmware-scan", "--input", str(self.media))
with self.assertRaises(runner.ValidationError):
self.plan("--timeout", "2", "capture", "--interface", "1", "--host", "192.0.2.1", "--duration", "20", "--output", str(self.root / "out.pcapng"))
def test_execution_uses_array_no_shell_finite_timeout_and_no_raw_error(self):
url = "rtsp://admin:TopSecret123@192.0.2.1/live?token=HiddenToken"
p = self.plan("probe", "--rtsp-env", "URL", env={"URL": url})
completed = subprocess.CompletedProcess(p.argv, 0, json.dumps({"streams": [{"codec_name": "h264", "width": 1920, "tags": {"url": url}}], "format": {"filename": url, "duration": "1.0"}}), url)
with mock.patch.object(runner, "_run_in_windows_job", return_value=completed) as launch:
result = runner.execute(p)
kwargs = launch.call_args.kwargs
self.assertFalse(kwargs["shell"])
self.assertGreater(kwargs["timeout"], 0)
self.assertIsInstance(launch.call_args.args[0], list)
self.assertEqual(result["execution_status"], "completed")
self.assertEqual(result["test_verdict"], "not_evaluated")
self.assertNotIn("TopSecret123", json.dumps(result))
self.assertNotIn("HiddenToken", json.dumps(result))
self.assertEqual(result["data"]["streams"][0]["width"], 1920)
def test_timeout_exception_never_echoes_raw_command_or_partial_output(self):
url = "rtsp://a:TopSecret123@192.0.2.1/live"
p = self.plan("probe", "--rtsp-env", "URL", env={"URL": url})
with mock.patch.object(runner, "_run_in_windows_job", side_effect=subprocess.TimeoutExpired(p.argv, 30, output=url, stderr=url)):
result = runner.execute(p)
self.assertEqual(result["execution_status"], "timed_out")
self.assertNotIn("TopSecret123", json.dumps(result))
def test_dry_run_does_not_launch_process(self):
profile = self.root / "profile.json"
profile.write_text(json.dumps(self.profile), encoding="utf-8")
output = io.StringIO()
with mock.patch.object(runner, "_run_in_windows_job") as launch, contextlib.redirect_stdout(output):
code = runner.main(["--profile", str(profile), "--dry-run", "scan", "--host", "192.0.2.1", "--ports", "80"])
self.assertEqual(code, 0)
self.assertFalse(launch.called)
self.assertEqual(json.loads(output.getvalue())["execution_status"], "planned")
def test_cli_errors_are_json_and_do_not_echo_secrets(self):
output = io.StringIO()
with contextlib.redirect_stdout(output):
code = runner.main(["probe", "--url", "rtsp://a:TopSecret123@192.0.2.1/live"])
self.assertEqual(code, 2)
self.assertEqual(json.loads(output.getvalue())["execution_status"], "invalid_arguments")
self.assertNotIn("TopSecret123", output.getvalue())
def test_doctor_without_version_flag_never_launches_tools(self):
with mock.patch.object(runner, "_run_in_windows_job") as launch, mock.patch.object(runner.shutil, "which", return_value=None):
result = runner.inventory({}, [], check_versions=False)
self.assertFalse(launch.called)
rows = {item["tool_id"]: item for item in result["tools"]}
self.assertIn(rows["tshark"]["availability"], ("missing", "package-only"))
self.assertEqual(rows["router"]["availability"], "manual-interface")
def test_inventory_preserves_mcp_and_orchestrator_routes(self):
with mock.patch.object(runner.shutil, "which", return_value=None):
rows = {item["tool_id"]: item for item in runner.inventory({}, [], False)["tools"]}
self.assertEqual(rows["artemis"]["interface"], "mcp")
self.assertEqual(rows["artemis"]["availability"], "mcp-check-required")
self.assertIn("mobile_workspace", rows["artemis"]["mcp_tools"])
self.assertEqual(rows["codex"]["interface"], "orchestrator")
self.assertEqual(rows["codex"]["availability"], "orchestrator-context")
def test_version_summary_uses_first_nonempty_line(self):
completed = subprocess.CompletedProcess([], 0, "\n \nNmap version 7.70\nMore detail\n", "")
with mock.patch.object(runner, "_run_in_windows_job", return_value=completed):
rows = {item["tool_id"]: item for item in runner.inventory(self.profile, [], True)["tools"]}
self.assertEqual(rows["nmap"]["version_check"]["first_line"], "Nmap version 7.70")
if __name__ == "__main__":
unittest.main()