335 lines
11 KiB
Python
335 lines
11 KiB
Python
"""Bounded, fixed, local dependency checks. No device or network operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gzip
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
MAX_CHECK_OUTPUT = 1024 * 1024
|
|
SERIAL_CHECK = (
|
|
"$ErrorActionPreference='Stop'; try { "
|
|
"$t=[System.IO.Ports.SerialPort]; "
|
|
"@{serial_port_type=$t.FullName;ports=@($t::GetPortNames())} | ConvertTo-Json -Compress "
|
|
"} catch { exit 1 }"
|
|
)
|
|
|
|
|
|
def process_lifecycle():
|
|
name = "_artemis_device_test_process_lifecycle"
|
|
if name not in sys.modules:
|
|
path = Path(__file__).resolve().parent / "skill/scripts/process_lifecycle.py"
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return sys.modules[name]
|
|
|
|
|
|
PROCESS = process_lifecycle()
|
|
|
|
|
|
def run_check(argv, directory, deadline, *, binwalk=False):
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0.15:
|
|
return {"status": "budget_exhausted"}, ""
|
|
env = os.environ.copy()
|
|
for key in ("FFREPORT", "PYTHONPATH", "PYTHONHOME"):
|
|
env.pop(key, None)
|
|
with tempfile.TemporaryDirectory(prefix="artemis-check-config-") as config:
|
|
if binwalk:
|
|
env["XDG_CONFIG_HOME"] = config
|
|
try:
|
|
completed = PROCESS.run(
|
|
argv,
|
|
shell=False,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
cwd=directory,
|
|
env=env,
|
|
timeout=min(3.0, remaining),
|
|
max_output_bytes=MAX_CHECK_OUTPUT,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return {"status": "timed_out", "error_code": "dependency_check_timeout"}, ""
|
|
except PROCESS.ManagedProcessError as exc:
|
|
return {
|
|
"status": "unknown" if exc.may_have_started else "failed",
|
|
"error_code": exc.error_code,
|
|
"may_have_started": exc.may_have_started,
|
|
}, ""
|
|
result = {
|
|
"status": "passed" if completed.returncode == 0 else "failed",
|
|
"returncode": completed.returncode,
|
|
}
|
|
if completed.returncode:
|
|
result["error_code"] = "dependency_returned_nonzero"
|
|
return result, ""
|
|
return result, completed.stdout.decode("utf-8", errors="replace")
|
|
|
|
|
|
def check_startup(tool_id, prefix, directory, deadline):
|
|
arguments = {
|
|
"ffprobe": ["-version"],
|
|
"ffmpeg": ["-version"],
|
|
"tshark": ["--version"],
|
|
"dumpcap": ["--version"],
|
|
"nmap": ["--version"],
|
|
"binwalk": ["--help"],
|
|
"mediainfo": ["--version"],
|
|
"powershell": [
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-Command",
|
|
"$PSVersionTable.PSVersion.ToString()",
|
|
],
|
|
}
|
|
result, output = run_check(
|
|
prefix + arguments[tool_id], directory, deadline, binwalk=tool_id == "binwalk"
|
|
)
|
|
result["check"] = "dll_load_and_required_api" if tool_id == "mediainfo" else "version_or_help"
|
|
if result["status"] == "passed":
|
|
# Exit 0 with empty output is only startup evidence, never functional evidence.
|
|
line = next((line.strip() for line in output.splitlines() if line.strip()), "")
|
|
# Do not echo arbitrary help text, local paths or environment values into receipts.
|
|
result["output_present"] = bool(line)
|
|
if tool_id == "mediainfo" and not line:
|
|
result.update(status="failed", error_code="mediainfo_version_api_empty")
|
|
return result
|
|
|
|
|
|
def riff_chunk(name, data):
|
|
return name + struct.pack("<I", len(data)) + data + (b"\x00" if len(data) % 2 else b"")
|
|
|
|
|
|
def video_sample():
|
|
"""A one-second AVI with two 16x16 uncompressed BGR frames, no external encoder."""
|
|
width, height, frames = 16, 16, 2
|
|
frame = b"\x20\x40\x80" * width * height
|
|
avih = struct.pack(
|
|
"<14I", 500000, len(frame) * 2, 0, 0x10, frames, 0, 1, len(frame), width, height, 0, 0, 0, 0
|
|
)
|
|
strh = struct.pack(
|
|
"<4s4sIHH8I4h",
|
|
b"vids",
|
|
b"DIB ",
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
2,
|
|
0,
|
|
frames,
|
|
len(frame),
|
|
0xFFFFFFFF,
|
|
0,
|
|
0,
|
|
0,
|
|
width,
|
|
height,
|
|
)
|
|
strf = struct.pack("<IiiHHIIiiII", 40, width, height, 1, 24, 0, len(frame), 0, 0, 0, 0)
|
|
stream = riff_chunk(b"LIST", b"strl" + riff_chunk(b"strh", strh) + riff_chunk(b"strf", strf))
|
|
header = riff_chunk(b"LIST", b"hdrl" + riff_chunk(b"avih", avih) + stream)
|
|
movie = riff_chunk(b"LIST", b"movi" + riff_chunk(b"00db", frame) * frames)
|
|
index = riff_chunk(
|
|
b"idx1",
|
|
b"".join(
|
|
struct.pack("<4sIII", b"00db", 0x10, 4 + i * (len(frame) + 8), len(frame))
|
|
for i in range(frames)
|
|
),
|
|
)
|
|
return riff_chunk(b"RIFF", b"AVI " + header + movie + index)
|
|
|
|
|
|
def pcap_sample():
|
|
payload = b"TEST"
|
|
ip = struct.pack(
|
|
"!BBHHHBBH4s4s", 0x45, 0, 32, 1, 0, 64, 17, 0, b"\xc0\x00\x02\x01", b"\xc0\x00\x02\x02"
|
|
)
|
|
checksum = sum(struct.unpack("!10H", ip))
|
|
checksum = (checksum & 0xFFFF) + (checksum >> 16)
|
|
ip = ip[:10] + struct.pack("!H", (~checksum) & 0xFFFF) + ip[12:]
|
|
packet = (
|
|
bytes.fromhex("0200000000020200000000010800")
|
|
+ ip
|
|
+ struct.pack("!4H", 12345, 54321, 12, 0)
|
|
+ payload
|
|
)
|
|
return (
|
|
struct.pack("<IHHIIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 1)
|
|
+ struct.pack("<4I", 1, 0, len(packet), len(packet))
|
|
+ packet
|
|
)
|
|
|
|
|
|
def check_function(tool_id, prefix, directory, deadline):
|
|
video = directory / "sample.avi"
|
|
video.write_bytes(video_sample())
|
|
capture = directory / "sample.pcap"
|
|
capture.write_bytes(pcap_sample())
|
|
firmware = directory / "sample.bin"
|
|
firmware.write_bytes(
|
|
b"\x00" * 256 + gzip.compress(b"ARTEMIS fixed dependency sample\n" * 8, mtime=0)
|
|
)
|
|
empty = directory / "empty-targets.txt"
|
|
empty.write_bytes(b"")
|
|
commands = {
|
|
"ffprobe": (
|
|
"local_video_metadata",
|
|
[
|
|
"-v",
|
|
"quiet",
|
|
"-protocol_whitelist",
|
|
"file",
|
|
"-show_entries",
|
|
"stream=codec_type,width,height",
|
|
"-of",
|
|
"json",
|
|
"-i",
|
|
str(video),
|
|
],
|
|
),
|
|
"ffmpeg": (
|
|
"local_video_decode",
|
|
[
|
|
"-v",
|
|
"error",
|
|
"-nostdin",
|
|
"-protocol_whitelist",
|
|
"file",
|
|
"-i",
|
|
str(video),
|
|
"-frames:v",
|
|
"2",
|
|
"-f",
|
|
"framehash",
|
|
"-hash",
|
|
"sha256",
|
|
"-",
|
|
],
|
|
),
|
|
"mediainfo": ("local_video_dll_metadata", ["--input", str(video)]),
|
|
"tshark": (
|
|
"local_pcap_fields",
|
|
[
|
|
"-n",
|
|
"-r",
|
|
str(capture),
|
|
"-c",
|
|
"1",
|
|
"-T",
|
|
"fields",
|
|
"-e",
|
|
"frame.number",
|
|
"-e",
|
|
"ip.src",
|
|
"-e",
|
|
"udp.dstport",
|
|
],
|
|
),
|
|
"dumpcap": ("interface_enumeration_only", ["-D"]),
|
|
"nmap": ("empty_target_list_xml_no_scan", ["-sL", "-n", "-iL", str(empty), "-oX", "-"]),
|
|
"binwalk": ("local_gzip_signature_no_extraction", [str(firmware)]),
|
|
"powershell": (
|
|
"serial_type_and_port_names_no_open",
|
|
["-NoLogo", "-NoProfile", "-NonInteractive", "-Command", SERIAL_CHECK],
|
|
),
|
|
}
|
|
name, tail = commands[tool_id]
|
|
result, output = run_check(prefix + tail, directory, deadline, binwalk=tool_id == "binwalk")
|
|
result["check"] = name
|
|
if result["status"] != "passed":
|
|
return result
|
|
valid = False
|
|
try:
|
|
if tool_id == "ffprobe":
|
|
valid = any(
|
|
row.get("codec_type") == "video"
|
|
and row.get("width") == 16
|
|
and row.get("height") == 16
|
|
for row in json.loads(output).get("streams", [])
|
|
)
|
|
elif tool_id == "ffmpeg":
|
|
valid = (
|
|
len(re.findall(r"(?m)^0,\s*\d+,\s*\d+,\s*\d+,\s*768,\s*[0-9a-f]{64}\s*$", output))
|
|
== 2
|
|
)
|
|
elif tool_id == "mediainfo":
|
|
valid = any(
|
|
row.get("Width") == "16" and row.get("Height") == "16"
|
|
for row in json.loads(output).get("video", [])
|
|
)
|
|
elif tool_id == "tshark":
|
|
valid = output.strip().split("\t") == ["1", "192.0.2.1", "54321"]
|
|
elif tool_id == "dumpcap":
|
|
count = len(re.findall(r"(?m)^\d+\.\s+\S", output))
|
|
valid = count > 0
|
|
result["interface_count"] = count
|
|
elif tool_id == "nmap":
|
|
root = ET.fromstring(output)
|
|
hosts = root.find("runstats/hosts")
|
|
valid = (
|
|
root.tag == "nmaprun"
|
|
and hosts is not None
|
|
and hosts.get("total") == "0"
|
|
and not root.findall("host")
|
|
)
|
|
elif tool_id == "binwalk":
|
|
valid = bool(re.search(r"(?i)gzip|gzip compressed", output)) and bool(
|
|
re.search(r"(?im)\b256\b|\b0x0*100\b", output)
|
|
)
|
|
elif tool_id == "powershell":
|
|
data = json.loads(output)
|
|
valid = data.get("serial_port_type") == "System.IO.Ports.SerialPort" and isinstance(
|
|
data.get("ports"), list
|
|
)
|
|
if valid:
|
|
result["port_count"] = len(data["ports"])
|
|
except (ValueError, AttributeError, TypeError, ET.ParseError):
|
|
valid = False
|
|
if not valid:
|
|
result.update(status="failed", error_code="fixed_local_sample_result_mismatch")
|
|
return result
|
|
|
|
|
|
def python_runtime(deadline, check_load):
|
|
modules = {
|
|
name: {"discovery": "found" if importlib.util.find_spec(name) else "missing"}
|
|
for name in ("mcp", "dotenv")
|
|
}
|
|
startup = {"status": "not_checked"}
|
|
if check_load:
|
|
with tempfile.TemporaryDirectory(prefix="artemis-python-check-") as directory:
|
|
startup, _ = run_check(
|
|
[
|
|
sys.executable,
|
|
"-I",
|
|
"-B",
|
|
"-c",
|
|
"import mcp, dotenv; print('dependency_imports_loaded')",
|
|
],
|
|
directory,
|
|
deadline,
|
|
)
|
|
return {
|
|
"python_bits": struct.calcsize("P") * 8,
|
|
"version": ".".join(map(str, sys.version_info[:3])),
|
|
"modules": modules,
|
|
"startup": startup,
|
|
"installer_installs_requirements": False,
|
|
}
|