From 5a2d4703ab5e78a99272c78526ed6ac3fe445a82 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Mon, 17 Aug 2026 16:17:05 -0600 Subject: [PATCH] Close the remaining review findings: injection, isolation, unbounded work 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 --- pyproject.toml | 3 +- src/mcqemu/launcher.py | 90 +++++++++++----- src/mcqemu/models.py | 1 + src/mcqemu/registry.py | 57 ++++++++++- src/mcqemu/tools/display.py | 48 +++++++-- src/mcqemu/tools/lifecycle.py | 32 +++++- src/mcqemu/tools/sandbox.py | 15 ++- tests/test_display.py | 2 +- tests/test_lifecycle.py | 15 ++- tests/test_reliability.py | 186 ++++++++++++++++++++++++++++++++++ tests/test_sandbox.py | 3 +- 11 files changed, 405 insertions(+), 47 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ae77aed..41ffff0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,8 @@ select = ["E", "F", "I", "UP", "B", "SIM", "ASYNC"] # ASYNC109: `timeout` params here are deliberate LLM-facing tool parameters. # ASYNC110: polling an external process's death has no event to await. # ASYNC240: pathlib use is microsecond stat/exists checks, not bulk I/O. -ignore = ["ASYNC109", "ASYNC110", "ASYNC240"] +# ASYNC230: the one blocking read is a bounded 256KB tail, not bulk I/O. +ignore = ["ASYNC109", "ASYNC110", "ASYNC230", "ASYNC240"] [tool.pytest.ini_options] asyncio_mode = "auto" diff --git a/src/mcqemu/launcher.py b/src/mcqemu/launcher.py index 3d2c894..6552253 100644 --- a/src/mcqemu/launcher.py +++ b/src/mcqemu/launcher.py @@ -15,6 +15,7 @@ from fastmcp.exceptions import ToolError from .config import DEFAULT_MACHINE, FIRMWARE_MAP, Config, host_arch, kvm_available from .models import VMConfig, VMRecord from .qemu_img import probe_format +from .registry import socket_is_live _FWD_RE = re.compile(r"^(\d{1,5}):(\d{1,5})$") @@ -58,25 +59,6 @@ def pick_free_port() -> int: return s.getsockname()[1] -def socket_is_live(path: Path) -> bool: - """True if something is still accepting connections on this unix socket. - - A leftover socket *file* is normal after a crash; a socket that still - accepts means a process owns this VM's name and must not be trampled. - """ - if not path.exists(): - return False - probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - try: - probe.settimeout(0.2) - probe.connect(str(path)) - return True - except OSError: - return False - finally: - probe.close() - - def resolve_port_forwards(forwards: list[str]) -> list[str]: """Turn user forward specs into concrete 'HOSTPORT:GUESTPORT' entries. @@ -86,6 +68,7 @@ def resolve_port_forwards(forwards: list[str]) -> list[str]: instead of a cryptic QEMU launch error. """ resolved: list[str] = [] + claimed: set[int] = set() for fwd in forwards: host, sep, guest = fwd.partition(":") if not sep: @@ -97,10 +80,17 @@ def resolve_port_forwards(forwards: list[str]) -> list[str]: ) if host in ("auto", "0"): port = pick_free_port() + while port in claimed: # each bind test closes before the next runs + port = pick_free_port() else: if not host.isdigit() or not 1 <= int(host) <= 65535: raise ToolError(f"Invalid host port in {fwd!r} — use a number or 'auto'.") port = int(host) + if port in claimed: + raise ToolError( + f"Host port {port} is requested twice in port_forwards; QEMU would " + "fail to launch. Use 'auto' for one of them." + ) try: with socket.socket() as s: s.bind(("", port)) @@ -109,10 +99,54 @@ def resolve_port_forwards(forwards: list[str]) -> list[str]: f"Host port {port} is already in use ({e.strerror}) — " f"pick another, or use 'auto:{guest}' to grab a free one." ) from e + claimed.add(port) resolved.append(f"{port}:{guest}") return resolved +def qopt(value: str | Path) -> str: + """Escape a value for QEMU's comma-separated option syntax. + + QEMU splits options on commas and reads a doubled comma as a literal one. + A path like 'data,readonly=on.qcow2' would otherwise inject an option. + """ + return str(value).replace(",", ",,") + + +# Flags that would let a "sandbox" reach the host filesystem, host devices, or +# spawn host processes. extra_args is an operator escape hatch, but an agent +# acting on untrusted input must not be able to open these doors. +_UNSAFE_EXTRA_ARGS = { + "-fsdev": "host filesystem passthrough", + "-virtfs": "host filesystem passthrough", + "-runas": "changes the QEMU process user", + "-monitor": "exposes the human monitor", + "-qmp": "would collide with the managed QMP socket", + "-pidfile": "would collide with the managed pidfile", + "-daemonize": "already applied by the launcher", +} + + +def check_extra_args(extra: list[str]) -> None: + """Reject the escape-hatch flags that break the sandbox's promises.""" + for i, token in enumerate(extra): + flag = token.split("=", 1)[0] + if flag in _UNSAFE_EXTRA_ARGS: + raise ToolError( + f"extra_args may not contain {flag!r} ({_UNSAFE_EXTRA_ARGS[flag]}). " + "Remove it, or start QEMU yourself and manage it with attach_vm." + ) + if flag == "-drive" or flag == "-blockdev": + value = extra[i + 1] if flag == token and i + 1 < len(extra) else token + if "/dev/" in value: + raise ToolError( + "extra_args may not attach host block devices (/dev/...). Use a " + "disk image created with image_create." + ) + if flag == "-chardev" and "spawn" in token: + raise ToolError("extra_args may not use a spawning chardev (runs host commands).") + + def build_cmdline( cfg: VMConfig, paths: VMPaths, @@ -144,19 +178,19 @@ def build_cmdline( if fw["mode"] == "pflash": args += [ "-drive", - f"if=pflash,format=raw,readonly=on,file={fw['code']}", + f"if=pflash,format=raw,readonly=on,file={qopt(fw['code'])}", "-drive", - f"if=pflash,format=raw,file={paths.uefi_vars}", + f"if=pflash,format=raw,file={qopt(paths.uefi_vars)}", ] else: args += ["-bios", fw["code"]] for disk in cfg.disks: fmt = disk_formats[disk] - args += ["-drive", f"file={disk},if=virtio,format={fmt}"] + args += ["-drive", f"file={qopt(disk)},if=virtio,format={fmt}"] if cfg.iso: - args += ["-drive", f"file={cfg.iso},media=cdrom,readonly=on"] + args += ["-drive", f"file={qopt(cfg.iso)},media=cdrom,readonly=on"] if cfg.disks: # Boot the installer once; subsequent boots hit the disk. args += ["-boot", "once=d"] @@ -165,6 +199,10 @@ def build_cmdline( args += ["-nic", "none"] else: netdev = "user,id=net0" + if cfg.restrict_net: + # Guest-initiated traffic is dropped: no internet, and no reaching + # host services on 10.0.2.2. Inbound hostfwd still works. + netdev += ",restrict=on" for fwd in cfg.port_forwards: m = _FWD_RE.match(fwd) if not m: @@ -178,11 +216,11 @@ def build_cmdline( "-display", "none", "-serial", - f"file:{paths.serial_log}", + f"file:{qopt(paths.serial_log)}", "-qmp", - f"unix:{paths.qmp_socket},server=on,wait=off", + f"unix:{qopt(paths.qmp_socket)},server=on,wait=off", "-chardev", - f"socket,id=qga0,path={paths.qga_socket},server=on,wait=off", + f"socket,id=qga0,path={qopt(paths.qga_socket)},server=on,wait=off", "-device", "virtio-serial", "-device", diff --git a/src/mcqemu/models.py b/src/mcqemu/models.py index 1801f31..563d5f3 100644 --- a/src/mcqemu/models.py +++ b/src/mcqemu/models.py @@ -20,6 +20,7 @@ class VMConfig: firmware: Literal["bios", "uefi"] = "bios" port_forwards: list[str] = field(default_factory=list) no_net: bool = False + restrict_net: bool = False extra_args: list[str] = field(default_factory=list) diff --git a/src/mcqemu/registry.py b/src/mcqemu/registry.py index b6323fd..e5a71ea 100644 --- a/src/mcqemu/registry.py +++ b/src/mcqemu/registry.py @@ -15,6 +15,7 @@ import contextlib import fcntl import json import os +import socket import time from collections.abc import Callable, Iterator from pathlib import Path @@ -25,6 +26,45 @@ from .models import VMRecord SCHEMA_VERSION = 1 +def socket_is_live(path: Path) -> bool: + """True if something is still accepting connections on this unix socket. + + A leftover socket *file* is normal after a crash; a socket that still + accepts means a process owns it and must not be trampled. + """ + if not path.exists(): + return False + probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + probe.settimeout(0.2) + probe.connect(str(path)) + return True + except OSError: + return False + finally: + probe.close() + + +def pid_matches_vm(pid: int | None, name: str) -> bool: + """False only when we can positively prove this PID is a *different* VM. + + Pairs with pid_alive: after a pidfile clobber a record can point at some + other QEMU, and acting on it (quit, SIGKILL) would hit the wrong machine. + Unreadable cmdline means 'cannot prove otherwise', so we do not contradict + the liveness check on that basis alone. + """ + try: + parts = Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0") + except OSError: + return True + if b"-name" not in parts: + return True + index = parts.index(b"-name") + if index + 1 >= len(parts): + return True + return parts[index + 1].decode(errors="replace") == name + + def pid_alive(pid: int | None) -> bool: """True iff the PID exists AND is actually a qemu-system process. @@ -207,11 +247,15 @@ class VMRegistry: def is_process_alive(self, record: VMRecord) -> bool: """Liveness for spawned VMs via pidfile; attached VMs are judged by - their QMP socket instead (we may not know their PID).""" + their QMP socket instead (we may not know their PID). + + A SIGKILLed QEMU leaves its socket file behind, so for attached VMs we + connect rather than just stat — a stale file must not read as alive. + """ if record.source == "attached" and record.pid is None: - return Path(record.qmp_socket).exists() + return socket_is_live(Path(record.qmp_socket)) self.refresh_pid(record) - return pid_alive(record.pid) + return pid_alive(record.pid) and pid_matches_vm(record.pid, record.name) def disks_in_use(self) -> dict[str, str]: """Map of absolute disk path -> VM name, for every live registered VM. @@ -224,6 +268,11 @@ class VMRegistry: for rec in self._vms.values(): if not self.is_process_alive(rec): continue - for disk in rec.config.get("disks", []): + disks = list(rec.config.get("disks", [])) + # A sandbox's base image is read by the running VM through the + # overlay's backing chain, so it is in use too. + if rec.config.get("sandbox_base"): + disks.append(rec.config["sandbox_base"]) + for disk in disks: used[str(Path(disk).expanduser().resolve())] = rec.name return used diff --git a/src/mcqemu/tools/display.py b/src/mcqemu/tools/display.py index dfe36c0..72d6c23 100644 --- a/src/mcqemu/tools/display.py +++ b/src/mcqemu/tools/display.py @@ -1,8 +1,11 @@ """See & drive: screenshots, keyboard/mouse injection, serial console.""" import asyncio +import contextlib import math +import os import struct +import uuid from pathlib import Path from typing import Literal @@ -16,6 +19,9 @@ from ..qmp import execute, qmp_session from ._common import require_vm _ABS_MAX = 32767 # QEMU absolute-pointer coordinate space +MAX_TYPE_CHARS = 4096 # each character is a separate QMP round trip +MAX_MOUSE_DELTA = 20000 # a full 4K screen traverse and then some +SERIAL_TAIL_BYTES = 256 * 1024 # PS/2-style relative buttons for HMP mouse_button (bitmask). _MOUSE_BUTTONS = {"left": 1, "right": 2, "middle": 4} @@ -31,14 +37,21 @@ def _png_dimensions(data: bytes) -> tuple[int, int]: async def _screendump(ctx: Context, name: str) -> bytes: state, record = require_vm(ctx, name) config: Config = state.config - dest = config.vm_state_dir(name) / "screenshot.png" + # Unique per call: a shared filename lets a concurrent screenshot swap the + # frame under vm_click between the capture and the size calculation, and + # leaves the guest's screen contents on disk afterwards. + dest = config.vm_state_dir(name) / f"screenshot-{uuid.uuid4().hex}.png" dest.parent.mkdir(parents=True, exist_ok=True) - async with qmp_session(record) as client: - await execute(client, "screendump", {"filename": str(dest), "format": "png"}) try: - return dest.read_bytes() - except OSError as e: - raise ToolError(f"screendump reported success but {dest} is unreadable: {e}") from e + async with qmp_session(record) as client: + await execute(client, "screendump", {"filename": str(dest), "format": "png"}) + try: + return dest.read_bytes() + except OSError as e: + raise ToolError(f"screendump reported success but {dest} is unreadable: {e}") from e + finally: + with contextlib.suppress(OSError): + dest.unlink(missing_ok=True) async def vm_screenshot(name: str, ctx: Context = None) -> Image: @@ -87,6 +100,12 @@ async def vm_type_text( printable ASCII plus tab and newline). Set enter=True to press Enter at the end — handy for shell commands at a console login or terminal.""" _, record = require_vm(ctx, name) + if len(text) > MAX_TYPE_CHARS: + raise ToolError( + f"text is {len(text)} characters; the limit is {MAX_TYPE_CHARS} because each " + "one is a separate keystroke round trip. Write large content with " + "guest_file_write instead." + ) presses: list[list[str]] = [] for char in text: try: @@ -185,9 +204,15 @@ async def vm_mouse_move( raise ToolError(f"Unsupported button {click!r}: use left, right, or middle.") if step < 1 or step > 120: raise ToolError("step must be between 1 and 120 (PS/2 deltas are small signed bytes).") + if abs(dx) > MAX_MOUSE_DELTA or abs(dy) > MAX_MOUSE_DELTA: + raise ToolError( + f"dx/dy must be within +/-{MAX_MOUSE_DELTA}: motion is sent in {step}px " + "packets, so a larger delta would issue an unbounded number of commands." + ) _, record = require_vm(ctx, name) + sx = sy = 0 homed_steps = 0 if home is not None: # Screen size bounds how far the corner can be; overshoot a little so @@ -244,10 +269,17 @@ async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) - path = Path(record.serial_log) if not path.exists(): raise ToolError(f"Serial log {path} does not exist yet.") - lines = path.read_text(errors="replace").splitlines() + # Read a window from the end: a chatty kernel console grows without bound + # and slurping the whole file would OOM the server managing every VM. + with path.open("rb") as handle: + size = handle.seek(0, os.SEEK_END) + handle.seek(max(0, size - SERIAL_TAIL_BYTES)) + window = handle.read().decode(errors="replace") + lines = window.splitlines() return { "name": name, - "total_lines": len(lines), + "log_bytes": size, + "read_from_end_bytes": min(size, SERIAL_TAIL_BYTES), "tail": "\n".join(lines[-tail_lines:]) if lines else "(serial log is empty)", } diff --git a/src/mcqemu/tools/lifecycle.py b/src/mcqemu/tools/lifecycle.py index d85b86c..5f5b444 100644 --- a/src/mcqemu/tools/lifecycle.py +++ b/src/mcqemu/tools/lifecycle.py @@ -8,7 +8,7 @@ from fastmcp import Context, FastMCP from fastmcp.exceptions import ToolError from ..config import resolve_image_path, valid_vm_name -from ..launcher import resolve_port_forwards, spawn_vm +from ..launcher import check_extra_args, resolve_port_forwards, spawn_vm from ..models import VMConfig from ..qmp import execute, qmp_session, recent_events, wait_for_event from ..registry import pid_alive @@ -26,6 +26,7 @@ async def launch_vm( firmware: Literal["bios", "uefi"] = "bios", port_forwards: list[str] | None = None, no_net: bool = False, + restrict_net: bool = False, extra_args: list[str] | None = None, ctx: Context = None, ) -> dict: @@ -39,12 +40,23 @@ async def launch_vm( acceleration is used automatically when the guest arch matches the host. The VM keeps running even if this MCP server restarts; stop it with stop_vm. + + restrict_net=True drops all guest-initiated traffic (no internet, and no + reaching services on the host) while keeping port_forwards working inbound + — use it when running untrusted software. no_net=True removes the NIC + entirely. + + extra_args passes raw flags to qemu-system-* and is an operator escape + hatch: only use values you wrote yourself, never values derived from + untrusted input. Flags that would breach VM isolation (host filesystem + passthrough, host block devices, spawning chardevs) are rejected. """ state = app(ctx) if not valid_vm_name(name): raise ToolError( f"Invalid VM name {name!r}: use letters, digits, '.', '_', '-' (max 48 chars)." ) + check_extra_args(extra_args or []) # Pick up records written by another instance (or another in-flight call) # before deciding this name is free. @@ -92,6 +104,7 @@ async def launch_vm( firmware=firmware, port_forwards=resolve_port_forwards(port_forwards or []), no_net=no_net, + restrict_net=restrict_net, extra_args=extra_args or [], ) record = await spawn_vm(cfg, state.config) @@ -190,13 +203,24 @@ async def attach_vm( ) if state.registry.get(name) is not None: raise ToolError(f"A VM named {name!r} is already registered. Use forget_vm first.") - if not Path(qmp_socket).exists(): - raise ToolError(f"No socket at {qmp_socket}. Is the QEMU process running with -qmp?") + sock = Path(qmp_socket).expanduser() + if not sock.exists(): + raise ToolError(f"No socket at {sock}. Is the QEMU process running with -qmp?") + if not sock.is_socket(): + raise ToolError( + f"{sock} is not a unix socket. Point qmp_socket at the path QEMU was given " + "in -qmp unix:,server=on." + ) from ..models import VMRecord + # Store the resolved path: a symlink swapped later must not redirect a quit. record = VMRecord( - name=name, source="attached", qmp_socket=qmp_socket, qga_socket=qga_socket, pid=pid + name=name, + source="attached", + qmp_socket=str(sock.resolve()), + qga_socket=str(Path(qga_socket).expanduser().resolve()) if qga_socket else None, + pid=pid, ) async with qmp_session(record) as client: status = await execute(client, "query-status") diff --git a/src/mcqemu/tools/sandbox.py b/src/mcqemu/tools/sandbox.py index efcbc33..5552fce 100644 --- a/src/mcqemu/tools/sandbox.py +++ b/src/mcqemu/tools/sandbox.py @@ -74,6 +74,7 @@ async def sandbox_vm( memory_mb: int = 2048, cpus: int = 2, port_forwards: list[str] | None = None, + allow_network: bool = False, wait_agent_s: int = 90, ctx: Context = None, ) -> dict: @@ -83,7 +84,17 @@ async def sandbox_vm( guest_file_* are immediately usable. By default a free host port is forwarded to guest port 22 (pass port_forwards=[] to disable, or your own list). If no `name` is given, sandbox / sandbox-2 / ... is chosen. - Tear everything down later with sandbox_destroy.""" + Tear everything down later with sandbox_destroy. + + Outbound networking is OFF by default: the guest cannot reach the internet + or any service on the host, which is what makes it a sandbox. Inbound port + forwards still work. Pass allow_network=True when the guest legitimately + needs to fetch packages. + + Note that the guest agent answers well before the guest finishes booting, + so this returns while services like networking are still starting. If a + command depends on one, wait for it inside the guest (e.g. poll + 'systemctl is-active NetworkManager') rather than assuming it is up.""" state = app(ctx) try: base = resolve_image_path(base_image) @@ -113,6 +124,7 @@ async def sandbox_vm( memory_mb=memory_mb, cpus=cpus, port_forwards=["auto:22"] if port_forwards is None else port_forwards, + restrict_net=not allow_network, ctx=ctx, ) except Exception: @@ -149,6 +161,7 @@ async def sandbox_vm( "sandbox": True, "overlay": str(overlay), "base_image": str(base), + "network": "outbound allowed" if allow_network else "outbound blocked", "guest_agent": agent, } ) diff --git a/tests/test_display.py b/tests/test_display.py index 4dacacc..4c8dac4 100644 --- a/tests/test_display.py +++ b/tests/test_display.py @@ -117,7 +117,7 @@ async def test_serial_read_tail(dirs, fake_qmp): data = result_data( await client.call_tool("vm_serial_read", {"name": "vm1", "tail_lines": 3}) ) - assert data["total_lines"] == 100 + assert data["log_bytes"] > 0 assert data["tail"] == "line97\nline98\nline99" diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 8ca4835..263949c 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -1,5 +1,7 @@ """Lifecycle tools with mocked spawn / fake QMP.""" +import socket + import pytest from fastmcp import Client from fastmcp.exceptions import ToolError @@ -140,9 +142,19 @@ async def test_attach_requires_existing_socket(client, tmp_path): ) +async def test_attach_rejects_a_regular_file(client, tmp_path): + """A path that exists but is not a socket cannot be a QMP endpoint.""" + plain = tmp_path / "not-a-socket" + plain.touch() + with pytest.raises(ToolError, match="not a unix socket"): + await client.call_tool("attach_vm", {"name": "ext", "qmp_socket": str(plain)}) + + async def test_attach_and_forget(dirs, fake_qmp, tmp_path): sock = tmp_path / "ext.sock" - sock.touch() + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(str(sock)) + listener.listen(1) FakeQMPClient.responses["query-status"] = {"status": "running"} async with Client(mcp) as client: data = result_data( @@ -153,6 +165,7 @@ async def test_attach_and_forget(dirs, fake_qmp, tmp_path): assert data["forgotten"] is True result = result_data(await client.call_tool("list_vms", {})) assert result["vms"] == [] + listener.close() async def test_forget_refuses_running_spawned_vm(dirs, all_pids_alive): diff --git a/tests/test_reliability.py b/tests/test_reliability.py index 2dc5d97..0c6619e 100644 --- a/tests/test_reliability.py +++ b/tests/test_reliability.py @@ -9,6 +9,7 @@ mid-call, two instances clobbering each other's records. import asyncio import json import socket +from pathlib import Path import pytest from fastmcp import Client @@ -259,3 +260,188 @@ def test_record_name_follows_the_registry_key(dirs): registry = VMRegistry(config_for(dirs)) registry.load() assert registry.get("truth").name == "truth" + + +# --- QEMU command line cannot be injected through paths --------------------- + + +def test_comma_in_disk_path_is_escaped(): + """A comma in a path would otherwise start a new QEMU option.""" + from mcqemu.launcher import build_cmdline + from mcqemu.models import VMConfig + from test_launcher import paths + + evil = "/vms/data,readonly=on,if=none.qcow2" + cfg = VMConfig(name="t", disks=[evil]) + args = build_cmdline(cfg, paths(), accel="kvm", disk_formats={evil: "qcow2"}) + drive = next(a for a in args if a.startswith("file=/vms/data")) + # Parse it the way QEMU does: split on single commas, un-double the rest. + options = dict(opt.split("=", 1) for opt in qemu_split(drive)) + assert options["file"] == evil, "the path must survive intact" + assert options["if"] == "virtio", "the embedded 'if=none' must not become an option" + assert set(options) == {"file", "if", "format"} + + +def qemu_split(option_string: str) -> list[str]: + """QEMU's option parser: ',,' is a literal comma, ',' separates options.""" + return [part.replace("\0", ",") for part in option_string.replace(",,", "\0").split(",")] + + +@pytest.mark.parametrize( + "bad", + [ + ["-fsdev", "local,path=/home,security_model=none,id=h"], + ["-virtfs", "local,path=/,mount_tag=root"], + ["-drive", "file=/dev/sda,format=raw"], + ["-runas", "root"], + ["-monitor", "stdio"], + ], +) +async def test_extra_args_rejects_isolation_breaks(client, bad): + with pytest.raises(ToolError, match="extra_args may not"): + await client.call_tool("launch_vm", {"name": "esc", "extra_args": bad}) + + +async def test_extra_args_allows_ordinary_flags(dirs, monkeypatch): + """The escape hatch still works for benign tuning flags.""" + from mcqemu.launcher import check_extra_args + + check_extra_args(["-vga", "std", "-rtc", "base=localtime"]) # must not raise + + +def test_duplicate_host_ports_are_rejected(): + from mcqemu.launcher import resolve_port_forwards + + with pytest.raises(ToolError, match="twice"): + resolve_port_forwards(["18991:22", "18991:23"]) + + +def test_auto_ports_do_not_collide_with_each_other(): + from mcqemu.launcher import resolve_port_forwards + + resolved = resolve_port_forwards(["auto:22", "auto:80", "auto:443"]) + hosts = [entry.split(":")[0] for entry in resolved] + assert len(set(hosts)) == 3 + + +def test_restrict_net_blocks_outbound_but_keeps_forwards(): + from mcqemu.launcher import build_cmdline + from mcqemu.models import VMConfig + from test_launcher import paths + + cfg = VMConfig(name="t", restrict_net=True, port_forwards=["2222:22"]) + netdev = next( + a + for a in build_cmdline(cfg, paths(), accel="kvm", disk_formats={}) + if a.startswith("user,id=net0") + ) + assert "restrict=on" in netdev + assert "hostfwd=tcp::2222-:22" in netdev + + +# --- bounded work per call -------------------------------------------------- + + +async def test_type_text_rejects_a_novel(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="limit is"): + await client.call_tool("vm_type_text", {"name": "vm1", "text": "x" * 5000}) + + +async def test_mouse_move_rejects_absurd_delta(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="within"): + await client.call_tool("vm_mouse_move", {"name": "vm1", "dx": 10**9}) + + +async def test_serial_read_only_reads_the_tail(dirs, fake_qmp): + """A multi-megabyte console log must not be slurped into memory.""" + log = dirs.state / "big-serial.log" + log.write_text("noise\n" * 200_000 + "FINAL LINE\n") + assert log.stat().st_size > 1_000_000 + write_registry(dirs, seeded_record(dirs, "vm1", serial_log=str(log))) + async with Client(mcp) as client: + data = result_data( + await client.call_tool("vm_serial_read", {"name": "vm1", "tail_lines": 1}) + ) + assert data["tail"] == "FINAL LINE" + assert data["log_bytes"] == log.stat().st_size + assert data["read_from_end_bytes"] < data["log_bytes"] + + +async def test_screenshots_do_not_share_a_filename(dirs, fake_qmp): + """A fixed path lets a concurrent capture swap the frame under vm_click.""" + written: list[str] = [] + + def capture(args): + written.append(args["filename"]) + with open(args["filename"], "wb") as f: + f.write(fake_png(64, 48)) + return {} + + from test_display import fake_png + + FakeQMPClient.responses["screendump"] = capture + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + await client.call_tool("vm_screenshot", {"name": "vm1"}) + await client.call_tool("vm_screenshot", {"name": "vm1"}) + assert len(set(written)) == 2, "each capture needs its own file" + assert not any(Path(p).exists() for p in written), "temp captures must be cleaned up" + + +# --- identity and liveness -------------------------------------------------- + + +def test_stale_socket_file_does_not_read_as_alive(dirs, tmp_path): + """A SIGKILLed QEMU leaves its socket behind; stat() alone would lie.""" + dead = tmp_path / "dead.sock" + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(str(dead)) + listener.listen(1) + registry = VMRegistry(config_for(dirs)) + registry.load() + record = VMRecord(name="ext", source="attached", qmp_socket=str(dead)) + assert registry.is_process_alive(record) is True + listener.close() # process gone, file remains + assert dead.exists() + assert registry.is_process_alive(record) is False + + +def test_pid_matching_rejects_a_different_vm(): + """After a pidfile clobber a record can point at another QEMU.""" + import os + + from mcqemu.registry import pid_matches_vm + + # Our own process has no -name in its cmdline, so nothing is proven. + assert pid_matches_vm(os.getpid(), "anything") is True + assert pid_matches_vm(2**22 + 999, "gone") is True # unreadable == unproven + + +def test_sandbox_base_counts_as_in_use(dirs): + """Resizing the base under a live overlay must be refused.""" + registry = VMRegistry(config_for(dirs)) + registry.load() + base = dirs.state / "base.qcow2" + base.write_bytes(b"x") + registry.add( + seeded_record( + dirs, + "sb", + source="attached", + pid=None, + qmp_socket=str(dirs.run / "live.sock"), + config={"disks": [], "sandbox_base": str(base)}, + ) + ) + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + (dirs.run).mkdir(exist_ok=True) + listener.bind(str(dirs.run / "live.sock")) + listener.listen(1) + try: + assert str(base.resolve()) in registry.disks_in_use() + finally: + listener.close() diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 2520bd5..6bcc62a 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -60,7 +60,7 @@ def base_image(tmp_path): def fake_launch(dirs, monkeypatch): """Replace the real spawn with one that registers a plausible record.""" - async def fake_launch_vm(name, disks, memory_mb, cpus, port_forwards, ctx): + async def fake_launch_vm(name, disks, memory_mb, cpus, port_forwards, ctx, **kwargs): from mcqemu.launcher import resolve_port_forwards from mcqemu.tools._common import app @@ -74,6 +74,7 @@ def fake_launch(dirs, monkeypatch): "status": "running", "accel": "kvm", "port_forwards": resolve_port_forwards(port_forwards), + "restrict_net": kwargs.get("restrict_net"), } monkeypatch.setattr("mcqemu.tools.sandbox.launch_vm", fake_launch_vm)