Ryan Malloy
5a2d4703ab
QEMU command line: - escape commas in every interpolated path (qopt); a path like 'data,readonly=on.qcow2' previously injected a drive option - reject extra_args flags that breach VM isolation (host filesystem passthrough, host block devices, spawning chardevs, -runas) and document the parameter as operator-only - detect duplicate host ports across port_forwards instead of failing at QEMU launch; auto ports no longer collide with each other Sandbox isolation: - sandbox_vm now blocks guest-initiated traffic by default (restrict=on), with allow_network=True to opt in. Verified end to end: with identical guest network state, a default sandbox reaches neither a host loopback service nor the internet, while allow_network=True reaches both - note in the docstring that the guest agent answers before the guest has finished booting Bounded work per call: - vm_serial_read seeks a 256KB window from the end instead of reading a console log that grows without bound into memory - cap vm_type_text length and vm_mouse_move deltas - screenshots get a unique filename and are cleaned up, so a concurrent capture cannot swap the frame under vm_click Identity and liveness: - attach_vm requires an actual unix socket and stores the resolved path - attached VMs are judged by connecting, not by a stat that a stale socket file would pass - refuse to act on a PID whose cmdline proves it is a different VM - a sandbox's base image counts as in use while its overlay is live - fix a latent NameError in vm_mouse_move's homing branch
199 lines
7.4 KiB
Python
199 lines
7.4 KiB
Python
"""See & drive tools with fake QMP; screendump writes a synthetic PNG."""
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from conftest import FakeQMPClient, result_data, write_registry
|
|
from mcqemu.server import mcp
|
|
from test_lifecycle import seeded_record
|
|
|
|
|
|
def fake_png(width: int, height: int) -> bytes:
|
|
"""Just enough PNG for signature + IHDR dimension parsing."""
|
|
return (
|
|
b"\x89PNG\r\n\x1a\n"
|
|
+ b"\x00\x00\x00\x0dIHDR"
|
|
+ struct.pack(">II", width, height)
|
|
+ b"\x08\x06\x00\x00\x00"
|
|
+ b"\x00" * 16
|
|
)
|
|
|
|
|
|
def arm_screendump(width=640, height=480):
|
|
def handler(args):
|
|
with open(args["filename"], "wb") as f:
|
|
f.write(fake_png(width, height))
|
|
return {}
|
|
|
|
FakeQMPClient.responses["screendump"] = handler
|
|
|
|
|
|
async def test_screenshot_returns_png_image(dirs, fake_qmp):
|
|
arm_screendump()
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("vm_screenshot", {"name": "vm1"})
|
|
block = result.content[0]
|
|
assert block.type == "image"
|
|
assert block.mimeType == "image/png"
|
|
call = next(a for c, a in FakeQMPClient.calls if c == "screendump")
|
|
assert call["format"] == "png"
|
|
|
|
|
|
async def test_send_keys_chords(dirs, fake_qmp):
|
|
FakeQMPClient.responses["send-key"] = {}
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"vm_send_keys", {"name": "vm1", "keys": ["ctrl-alt-f2", "enter"], "delay_ms": 0}
|
|
)
|
|
)
|
|
assert data["keys_sent"] == 2
|
|
sent = [a["keys"] for c, a in FakeQMPClient.calls if c == "send-key"]
|
|
assert sent[0] == [
|
|
{"type": "qcode", "data": "ctrl"},
|
|
{"type": "qcode", "data": "alt"},
|
|
{"type": "qcode", "data": "f2"},
|
|
]
|
|
assert sent[1] == [{"type": "qcode", "data": "ret"}]
|
|
|
|
|
|
async def test_send_keys_unknown_key(dirs, fake_qmp):
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError, match="Unknown key"):
|
|
await client.call_tool("vm_send_keys", {"name": "vm1", "keys": ["warpdrive"]})
|
|
|
|
|
|
async def test_type_text_with_shift_and_enter(dirs, fake_qmp):
|
|
FakeQMPClient.responses["send-key"] = {}
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"vm_type_text", {"name": "vm1", "text": "Hi!", "enter": True, "delay_ms": 0}
|
|
)
|
|
)
|
|
assert data["characters_typed"] == 4 # H, i, !, Enter
|
|
sent = [a["keys"] for c, a in FakeQMPClient.calls if c == "send-key"]
|
|
assert sent[0] == [{"type": "qcode", "data": "shift"}, {"type": "qcode", "data": "h"}]
|
|
assert sent[1] == [{"type": "qcode", "data": "i"}]
|
|
assert sent[2] == [{"type": "qcode", "data": "shift"}, {"type": "qcode", "data": "1"}]
|
|
assert sent[3] == [{"type": "qcode", "data": "ret"}]
|
|
|
|
|
|
async def test_click_scales_to_abs_space(dirs, fake_qmp):
|
|
arm_screendump(width=640, height=480)
|
|
FakeQMPClient.responses["input-send-event"] = {}
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(await client.call_tool("vm_click", {"name": "vm1", "x": 320, "y": 240}))
|
|
assert "left at (320, 240)" in data["clicked"]
|
|
events = [a["events"] for c, a in FakeQMPClient.calls if c == "input-send-event"]
|
|
move = events[0]
|
|
assert move[0]["data"] == {"axis": "x", "value": int(320 * 32767 / 639)}
|
|
assert move[1]["data"] == {"axis": "y", "value": int(240 * 32767 / 479)}
|
|
assert events[1][0]["data"] == {"down": True, "button": "left"}
|
|
assert events[2][0]["data"] == {"down": False, "button": "left"}
|
|
|
|
|
|
async def test_click_outside_display_rejected(dirs, fake_qmp):
|
|
arm_screendump(width=640, height=480)
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError, match="outside"):
|
|
await client.call_tool("vm_click", {"name": "vm1", "x": 9999, "y": 10})
|
|
|
|
|
|
async def test_serial_read_tail(dirs, fake_qmp):
|
|
record = seeded_record(dirs, "vm1", serial_log=str(dirs.state / "serial.log"))
|
|
(dirs.state / "serial.log").write_text("\n".join(f"line{i}" for i in range(100)))
|
|
write_registry(dirs, record)
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool("vm_serial_read", {"name": "vm1", "tail_lines": 3})
|
|
)
|
|
assert data["log_bytes"] > 0
|
|
assert data["tail"] == "line97\nline98\nline99"
|
|
|
|
|
|
def hmp_calls():
|
|
return [a["command-line"] for c, a in FakeQMPClient.calls if c == "human-monitor-command"]
|
|
|
|
|
|
async def test_mouse_move_chunks_deltas(dirs, fake_qmp):
|
|
FakeQMPClient.responses["human-monitor-command"] = ""
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"vm_mouse_move", {"name": "vm1", "dx": 70, "dy": -75, "step": 32}
|
|
)
|
|
)
|
|
assert data["moved"] == [70, -75]
|
|
assert hmp_calls() == [
|
|
"mouse_move 32 0",
|
|
"mouse_move 32 0",
|
|
"mouse_move 6 0",
|
|
"mouse_move 0 -32",
|
|
"mouse_move 0 -32",
|
|
"mouse_move 0 -11",
|
|
]
|
|
|
|
|
|
async def test_mouse_move_home_pins_to_corner(dirs, fake_qmp):
|
|
arm_screendump(width=320, height=200)
|
|
FakeQMPClient.responses["human-monitor-command"] = ""
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"vm_mouse_move", {"name": "vm1", "home": "bottom-right", "step": 100}
|
|
)
|
|
)
|
|
assert data["homed"] == "bottom-right"
|
|
moves = hmp_calls()
|
|
# ceil(320/100) + 2 = 6 homing steps, all toward +x/+y
|
|
assert moves == ["mouse_move 100 100"] * 6
|
|
|
|
|
|
async def test_mouse_move_home_top_left_direction(dirs, fake_qmp):
|
|
arm_screendump(width=100, height=100)
|
|
FakeQMPClient.responses["human-monitor-command"] = ""
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
await client.call_tool("vm_mouse_move", {"name": "vm1", "home": "top-left", "step": 50})
|
|
assert set(hmp_calls()) == {"mouse_move -50 -50"}
|
|
|
|
|
|
async def test_mouse_move_click_sequence(dirs, fake_qmp):
|
|
FakeQMPClient.responses["human-monitor-command"] = ""
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
data = result_data(
|
|
await client.call_tool(
|
|
"vm_mouse_move", {"name": "vm1", "click": "left", "double": True}
|
|
)
|
|
)
|
|
assert data["clicked"] == "left double"
|
|
assert hmp_calls() == ["mouse_button 1", "mouse_button 0"] * 2
|
|
|
|
|
|
async def test_mouse_move_right_click(dirs, fake_qmp):
|
|
FakeQMPClient.responses["human-monitor-command"] = ""
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
await client.call_tool("vm_mouse_move", {"name": "vm1", "click": "right"})
|
|
assert hmp_calls() == ["mouse_button 2", "mouse_button 0"]
|
|
|
|
|
|
async def test_mouse_move_rejects_bad_step(dirs, fake_qmp):
|
|
write_registry(dirs, seeded_record(dirs, "vm1"))
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError, match="step"):
|
|
await client.call_tool("vm_mouse_move", {"name": "vm1", "dx": 10, "step": 500})
|