354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Portable dependency discovery around the byte-preserved Skill runner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
|
|
PACKAGE = Path(__file__).resolve().parent
|
|
# -I excludes even the script directory. Only add this installed package, never cwd.
|
|
sys.path.insert(0, str(PACKAGE))
|
|
from dependency_checks import check_function, check_startup, python_runtime
|
|
|
|
|
|
def upstream_runner():
|
|
name = "artemis_device_test_upstream_runner"
|
|
spec = importlib.util.spec_from_file_location(
|
|
name, PACKAGE / "skill" / "scripts" / "tool_runner.py"
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
UPSTREAM = upstream_runner()
|
|
|
|
|
|
class DependencyError(ValueError):
|
|
"""Only stable, public error codes belong here."""
|
|
|
|
|
|
def dll_machine(path: Path) -> int:
|
|
"""Read just the DOS/PE headers; discovering a DLL never loads it."""
|
|
try:
|
|
with path.open("rb") as stream:
|
|
header = stream.read(64)
|
|
if len(header) != 64 or header[:2] != b"MZ":
|
|
raise ValueError
|
|
offset = struct.unpack_from("<I", header, 60)[0]
|
|
if not 64 <= offset <= 1024 * 1024:
|
|
raise ValueError
|
|
stream.seek(offset)
|
|
pe = stream.read(24)
|
|
if len(pe) != 24 or pe[:4] != b"PE\x00\x00":
|
|
raise ValueError
|
|
return struct.unpack_from("<H", pe, 4)[0]
|
|
except (OSError, ValueError, struct.error):
|
|
raise DependencyError("mediainfo_invalid_or_unreadable_pe") from None
|
|
|
|
|
|
def validate_library(path: Path) -> None:
|
|
machine = dll_machine(path)
|
|
expected = 0x8664 if struct.calcsize("P") == 8 else 0x14C
|
|
if machine != expected:
|
|
raise DependencyError("mediainfo_architecture_mismatch")
|
|
|
|
|
|
def path_directories():
|
|
seen = set()
|
|
for value in os.environ.get("PATH", "").split(os.pathsep):
|
|
value = value.strip().strip('"')
|
|
if not value or value.startswith(("\\\\", "//")):
|
|
continue
|
|
path = Path(value)
|
|
# Ignore empty/relative PATH entries: they must not search the working directory.
|
|
if not path.is_absolute() or not path.is_dir():
|
|
continue
|
|
resolved = path.resolve()
|
|
if resolved not in seen:
|
|
seen.add(resolved)
|
|
yield resolved
|
|
|
|
|
|
def media_library(profile, roots):
|
|
configured = profile.get("libraries", {}).get("mediainfo")
|
|
if configured:
|
|
path = Path(UPSTREAM.configured_local_path(configured, "mediainfo", r"MediaInfo\.dll"))
|
|
validate_library(path)
|
|
return str(path), "profile", []
|
|
rejected = []
|
|
candidates = [
|
|
(path, "tool_root")
|
|
for path in UPSTREAM.root_candidates(roots)
|
|
if path.name.lower() == "mediainfo.dll" and path.is_file()
|
|
]
|
|
candidates.extend((directory / "MediaInfo.dll", "path") for directory in path_directories())
|
|
seen = set()
|
|
for path, source in candidates:
|
|
if not path.is_file() or path.resolve() in seen:
|
|
continue
|
|
seen.add(path.resolve())
|
|
try:
|
|
validate_library(path)
|
|
except DependencyError as exc:
|
|
rejected.append({"path": str(path), "error_code": str(exc)})
|
|
continue
|
|
return str(path.resolve()), source, rejected
|
|
if rejected:
|
|
raise DependencyError("mediainfo_no_matching_architecture_or_valid_pe")
|
|
raise UPSTREAM.ToolMissing("mediainfo")
|
|
|
|
|
|
def resolve_dependency(tool_id, profile, roots):
|
|
"""Resolve one dependency only; an unrelated broken dependency cannot block it."""
|
|
effective = copy.deepcopy(profile)
|
|
rejected = []
|
|
if tool_id == "mediainfo":
|
|
path, source, rejected = media_library(profile, roots)
|
|
effective.setdefault("libraries", {})["mediainfo"] = path
|
|
elif (
|
|
tool_id == "binwalk"
|
|
and not profile.get("runtimes", {}).get("binwalk_python")
|
|
and not profile.get("tools", {}).get("binwalk")
|
|
):
|
|
# One documented opt-in directory convention, never an arbitrary Python on PATH.
|
|
for root in roots:
|
|
runtime = Path(root) / "binwalk-python" / "Scripts" / "python.exe"
|
|
if runtime.is_file():
|
|
effective.setdefault("runtimes", {})["binwalk_python"] = str(runtime.resolve())
|
|
break
|
|
prefix = UPSTREAM.command_prefix(tool_id, effective, roots)
|
|
if tool_id != "mediainfo":
|
|
path = prefix[0]
|
|
if profile.get("runtimes", {}).get("binwalk_python") and tool_id == "binwalk":
|
|
source = "profile.runtime"
|
|
elif profile.get("tools", {}).get(tool_id):
|
|
source = "profile"
|
|
elif tool_id == "binwalk" and effective.get("runtimes", {}).get("binwalk_python"):
|
|
source = "tool_root.binwalk-python"
|
|
else:
|
|
source = (
|
|
"tool_root"
|
|
if any(Path(path).is_relative_to(Path(root).resolve()) for root in roots)
|
|
else "path"
|
|
)
|
|
return (
|
|
effective,
|
|
prefix,
|
|
{"status": "found", "source": source, "path": path, "rejected": rejected},
|
|
)
|
|
|
|
|
|
CAPABILITIES = {
|
|
"probe_media": (
|
|
"ffprobe",
|
|
["RTSP connectivity and business expectations require task evidence"],
|
|
),
|
|
"media_info": ("mediainfo", []),
|
|
"analyze_pcap": ("tshark", []),
|
|
"capture_packets": (
|
|
"dumpcap",
|
|
["selected interface, Npcap and capture permission; no capture attempted"],
|
|
),
|
|
"scan_ports": ("nmap", ["target connectivity and requested port states; no scan attempted"]),
|
|
"scan_firmware": (
|
|
"binwalk",
|
|
["signature detection is not firmware compatibility or encryption acceptance"],
|
|
),
|
|
"capture_serial": (
|
|
"powershell",
|
|
["adapter driver, selected COM, baud, permission and exclusive access; no port opened"],
|
|
),
|
|
}
|
|
|
|
|
|
def capability_status(tools):
|
|
by_id = {row["tool_id"]: row for row in tools}
|
|
result = {}
|
|
for capability, (tool_id, conditions) in CAPABILITIES.items():
|
|
row = by_id[tool_id]
|
|
discovery, startup, function = (
|
|
row[key]["status"] for key in ("discovery", "startup", "functional")
|
|
)
|
|
blocked = (
|
|
discovery != "found"
|
|
or "failed" in {startup, function}
|
|
or "timed_out" in {startup, function}
|
|
)
|
|
result[capability] = {
|
|
"dependency": tool_id,
|
|
"status": "blocked"
|
|
if blocked
|
|
else "local_verified"
|
|
if function == "passed"
|
|
else "available_unverified",
|
|
"external_conditions": conditions,
|
|
}
|
|
return result
|
|
|
|
|
|
def doctor(profile, roots, *, check_versions=False, check_functional=False, timeout=42):
|
|
started = time.monotonic()
|
|
deadline = started + max(0.1, timeout - 1)
|
|
UPSTREAM.validate_profile(profile)
|
|
rows, prefixes = [], {}
|
|
catalog = json.loads(UPSTREAM.CATALOG.read_text("utf-8"))
|
|
for entry in catalog["tools"]:
|
|
row = dict(entry)
|
|
tool_id = entry["tool_id"]
|
|
if tool_id == "mediainfo":
|
|
row["capability"] = (
|
|
"通过 profile/统一目录/PATH 发现匹配位数的本地 MediaInfo.dll；独立子进程固定 API 和白名单字段，不使用 GUI/CLI 替代。"
|
|
)
|
|
if tool_id not in UPSTREAM.EXECUTABLES and tool_id != "mediainfo":
|
|
row["availability"] = {
|
|
"mcp": "mcp-check-required",
|
|
"orchestrator": "orchestrator-context",
|
|
}.get(entry.get("interface"), "manual-interface")
|
|
rows.append(row)
|
|
continue
|
|
row.update(startup={"status": "not_checked"}, functional={"status": "not_checked"})
|
|
try:
|
|
_, prefix, discovery = resolve_dependency(tool_id, profile, roots)
|
|
prefixes[tool_id] = prefix
|
|
row["discovery"] = discovery
|
|
row["availability"] = (
|
|
"library-found"
|
|
if tool_id == "mediainfo"
|
|
else "python-module-configured"
|
|
if tool_id == "binwalk" and len(prefix) > 1
|
|
else "executable-found"
|
|
)
|
|
except UPSTREAM.ToolMissing:
|
|
row.update(
|
|
availability="missing",
|
|
discovery={"status": "missing", "error_code": "dependency_not_found"},
|
|
)
|
|
except (UPSTREAM.ValidationError, DependencyError) as exc:
|
|
row.update(
|
|
availability="invalid", discovery={"status": "invalid", "error_code": str(exc)}
|
|
)
|
|
except OSError:
|
|
row.update(
|
|
availability="unavailable",
|
|
discovery={"status": "unavailable", "error_code": "dependency_path_unreadable"},
|
|
)
|
|
rows.append(row)
|
|
runtime = python_runtime(deadline, check_versions or check_functional)
|
|
with tempfile.TemporaryDirectory(prefix="artemis-dependency-check-") as temporary:
|
|
directory = Path(temporary)
|
|
# Complete startup checks first, so a slow functional check cannot hide discovery/load results.
|
|
if check_versions or check_functional:
|
|
for row in rows:
|
|
tool_id = row["tool_id"]
|
|
if tool_id in prefixes:
|
|
row["startup"] = check_startup(tool_id, prefixes[tool_id], directory, deadline)
|
|
if check_functional:
|
|
for row in rows:
|
|
tool_id = row["tool_id"]
|
|
if tool_id in prefixes:
|
|
row["functional"] = (
|
|
check_function(tool_id, prefixes[tool_id], directory, deadline)
|
|
if row["startup"]["status"] == "passed"
|
|
else {"status": "not_checked", "reason": "startup_not_verified"}
|
|
)
|
|
return {
|
|
"execution_status": "completed",
|
|
"test_verdict": "not_evaluated",
|
|
"tools": rows,
|
|
"capabilities": capability_status(rows),
|
|
"python_runtime": runtime,
|
|
"checks": {
|
|
"startup_requested": check_versions or check_functional,
|
|
"functional_requested": check_functional,
|
|
},
|
|
"external_conditions": {
|
|
"capture_driver": "interface enumeration does not prove packet capture permission",
|
|
"serial_driver": "port enumeration does not prove adapter driver or communication",
|
|
"gui_and_websites": "require actual application/window, authorized session and permissions",
|
|
"android": "requires Artemis device checks in the current task",
|
|
},
|
|
"device_actions_executed": False,
|
|
"catalog_evidence_is_historical": True,
|
|
"elapsed_seconds": round(time.monotonic() - started, 3),
|
|
"note": "Independent local checks; no scan, capture, serial open, download, installation or business verdict. PATH is inherited from the Artemis host process.",
|
|
}
|
|
|
|
|
|
def main(argv=None):
|
|
arguments = list(sys.argv[1:] if argv is None else argv)
|
|
functional = "--check-functional" in arguments
|
|
if functional:
|
|
arguments.remove("--check-functional")
|
|
try:
|
|
args = UPSTREAM.parse_args(arguments)
|
|
if functional and args.command != "doctor":
|
|
raise UPSTREAM.ValidationError("functional_check_only_supported_for_doctor")
|
|
profile = UPSTREAM.load_profile(args.profile)
|
|
if args.command in {"list", "doctor"}:
|
|
result = doctor(
|
|
profile,
|
|
args.tool_root,
|
|
check_versions=bool(getattr(args, "check_versions", False) and not args.dry_run),
|
|
check_functional=functional and not args.dry_run,
|
|
timeout=UPSTREAM.bounded_int(
|
|
args.timeout or "42", 1, 3600, "timeout_must_be_1_to_3600_seconds"
|
|
),
|
|
)
|
|
else:
|
|
tool_id = {
|
|
"probe": "ffprobe",
|
|
"media-info": "mediainfo",
|
|
"analyze-pcap": "tshark",
|
|
"capture": "dumpcap",
|
|
"scan": "nmap",
|
|
"firmware-scan": "binwalk",
|
|
"serial": "powershell",
|
|
}[args.command]
|
|
effective, _, discovery = resolve_dependency(tool_id, profile, args.tool_root)
|
|
plan = UPSTREAM.build_plan(args, effective, os.environ)
|
|
result = (
|
|
{"execution_status": "planned", "test_verdict": "not_evaluated", **plan.public()}
|
|
if args.dry_run
|
|
else UPSTREAM.execute(plan)
|
|
)
|
|
result["dependency_discovery"] = discovery
|
|
code = 0 if result["execution_status"] in {"planned", "completed"} else 1
|
|
except UPSTREAM.ToolMissing as exc:
|
|
result = {
|
|
"execution_status": "tool_missing",
|
|
"test_verdict": "not_evaluated",
|
|
"tool_id": str(exc),
|
|
"error_code": "prepare_dependency_in_path_tool_root_or_explicit_profile",
|
|
}
|
|
code = 3
|
|
except (UPSTREAM.ValidationError, DependencyError) as exc:
|
|
result = {
|
|
"execution_status": "invalid_arguments",
|
|
"test_verdict": "not_evaluated",
|
|
"error_code": str(exc),
|
|
}
|
|
code = 2
|
|
except (OSError, ValueError):
|
|
result = {
|
|
"execution_status": "configuration_error",
|
|
"test_verdict": "not_evaluated",
|
|
"error_code": "local_file_unreadable_or_invalid",
|
|
}
|
|
code = 2
|
|
print(json.dumps(result, ensure_ascii=True, indent=2))
|
|
return code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|