Add auto/collision-checked port forwards and one-shot sandbox_vm/sandbox_destroy

src/mcqemu/launcher.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import re
 import shutil
+import socket
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
@@ -48,6 +49,48 @@ def pick_accel(arch: str) -> str:
     return "kvm" if arch == host_arch() and kvm_available() else "tcg"
 
 
+def pick_free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def resolve_port_forwards(forwards: list[str]) -> list[str]:
+    """Turn user forward specs into concrete 'HOSTPORT:GUESTPORT' entries.
+
+    Accepted forms: "2222:22" (explicit, verified free), "auto:22" (a free
+    host port is picked), "22" (shorthand for auto:22). Explicit ports are
+    bind-tested up front so a collision fails here with a clear message
+    instead of a cryptic QEMU launch error.
+    """
+    resolved: list[str] = []
+    for fwd in forwards:
+        host, sep, guest = fwd.partition(":")
+        if not sep:
+            host, guest = "auto", host
+        if not guest.isdigit() or not 1 <= int(guest) <= 65535:
+            raise ToolError(
+                f"Invalid port forward {fwd!r} — use 'HOST:GUEST' (e.g. '2222:22'), "
+                "'auto:GUEST' to pick a free host port, or just 'GUEST'."
+            )
+        if host in ("auto", "0"):
+            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)
+            try:
+                with socket.socket() as s:
+                    s.bind(("", port))
+            except OSError as e:
+                raise ToolError(
+                    f"Host port {port} is already in use ({e.strerror}) — "
+                    f"pick another, or use 'auto:{guest}' to grab a free one."
+                ) from e
+        resolved.append(f"{port}:{guest}")
+    return resolved
+
+
 def build_cmdline(
     cfg: VMConfig,
     paths: VMPaths,

src/mcqemu/tools/__init__.py

@@ -2,9 +2,9 @@
 
 from fastmcp import FastMCP
 
-from . import display, guest, images, lifecycle, query, snapshots
+from . import display, guest, images, lifecycle, query, sandbox, snapshots
 
 
 def register_all(mcp: FastMCP) -> None:
-    for module in (lifecycle, query, snapshots, images, guest, display):
+    for module in (lifecycle, query, snapshots, images, guest, display, sandbox):
         module.register(mcp)

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 spawn_vm
+from ..launcher import resolve_port_forwards, spawn_vm
 from ..models import VMConfig
 from ..qmp import execute, qmp_session, wait_for_event
 from ..registry import pid_alive
@@ -33,10 +33,12 @@ async def launch_vm(
 
     Use `disks` for existing image files (create them first with image_create);
     use `iso` to boot an installer or live CD. With both, the VM boots the ISO
-    once then the disk afterwards. `port_forwards` entries like "2222:22" map
-    host port 2222 to guest port 22 (user-mode networking). KVM 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.
+    once then the disk afterwards. `port_forwards` maps host ports to guest
+    ports (user-mode networking): "2222:22" is explicit, "auto:22" (or just
+    "22") picks a free host port — the result reports what was chosen. KVM
+    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.
     """
     state = app(ctx)
     if not valid_vm_name(name):
@@ -79,7 +81,7 @@ async def launch_vm(
         disks=resolved_disks,
         iso=resolved_iso,
         firmware=firmware,
-        port_forwards=port_forwards or [],
+        port_forwards=resolve_port_forwards(port_forwards or []),
         no_net=no_net,
         extra_args=extra_args or [],
     )

src/mcqemu/tools/sandbox.py

@@ -0,0 +1,148 @@
+"""One-shot disposable sandboxes: overlay + launch + wait-for-agent, and
+the matching destroy that cleans up everything the sandbox created."""
+
+import asyncio
+import shutil
+from pathlib import Path
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..config import resolve_image_path, valid_vm_name
+from ..qemu_img import probe_format, run_qemu_img
+from ..qga import qga_session
+from ..qmp import execute, qmp_session
+from ..registry import pid_alive
+from ._common import app, require_vm
+from .lifecycle import launch_vm
+
+
+async def _agent_responding(record) -> bool:
+    try:
+        async with qga_session(record):
+            return True
+    except ToolError:
+        return False
+
+
+def _auto_name(taken: list[str]) -> str:
+    if "sandbox" not in taken:
+        return "sandbox"
+    i = 2
+    while f"sandbox-{i}" in taken:
+        i += 1
+    return f"sandbox-{i}"
+
+
+async def sandbox_vm(
+    base_image: str,
+    name: str | None = None,
+    memory_mb: int = 2048,
+    cpus: int = 2,
+    port_forwards: list[str] | None = None,
+    wait_agent_s: int = 90,
+    ctx: Context = None,
+) -> dict:
+    """Spin up a disposable sandbox VM from a base disk image, in one call:
+    creates a copy-on-write overlay (the base image is never modified),
+    launches the VM, and waits for the guest agent to come up so guest_exec /
+    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."""
+    state = app(ctx)
+    try:
+        base = resolve_image_path(base_image)
+    except ValueError as e:
+        raise ToolError(f"Bad base image: {e}") from e
+
+    if name is None:
+        name = _auto_name(state.registry.names())
+    elif not valid_vm_name(name):
+        raise ToolError(
+            f"Invalid VM name {name!r}: use letters, digits, '.', '_', '-' (max 48 chars)."
+        )
+
+    overlay_dir = state.config.vm_state_dir(name)
+    overlay_dir.mkdir(parents=True, exist_ok=True)
+    overlay = overlay_dir / "overlay.qcow2"
+    base_fmt = await probe_format(str(base))
+    # Recreate the overlay fresh each time — a stale one from a dead sandbox
+    # of the same name would resurrect old state.
+    overlay.unlink(missing_ok=True)
+    await run_qemu_img("create", "-f", "qcow2", "-b", str(base), "-F", base_fmt, str(overlay))
+
+    result = await launch_vm(
+        name=name,
+        disks=[str(overlay)],
+        memory_mb=memory_mb,
+        cpus=cpus,
+        port_forwards=["auto:22"] if port_forwards is None else port_forwards,
+        ctx=ctx,
+    )
+
+    record = state.registry.get(name)
+    record.config["sandbox_overlay"] = str(overlay)
+    record.config["sandbox_base"] = str(base)
+    state.registry.save()
+
+    agent = "unavailable"
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + wait_agent_s
+    while loop.time() < deadline:
+        if await _agent_responding(record):
+            agent = "responding"
+            break
+        await asyncio.sleep(2)
+
+    result.update(
+        {
+            "sandbox": True,
+            "overlay": str(overlay),
+            "base_image": str(base),
+            "guest_agent": agent,
+        }
+    )
+    if agent != "responding":
+        result["agent_hint"] = (
+            f"Guest agent did not answer within {wait_agent_s}s — the base image "
+            "may not have qemu-guest-agent installed, or the guest is still "
+            "booting. guest_* tools will fail until it responds (retry guest_ping)."
+        )
+    return result
+
+
+async def sandbox_destroy(name: str, ctx: Context = None) -> dict:
+    """Destroy a sandbox created with sandbox_vm: force-stop the VM, remove it
+    from the registry, and delete its overlay disk and logs. The base image is
+    untouched. Refuses to operate on VMs that were not created by sandbox_vm —
+    use stop_vm / forget_vm for those (they never delete disks)."""
+    state, record = require_vm(ctx, name)
+    overlay = record.config.get("sandbox_overlay")
+    if not overlay:
+        raise ToolError(
+            f"VM {name!r} was not created by sandbox_vm; refusing to delete its "
+            "disks. Use stop_vm and forget_vm instead."
+        )
+
+    if state.registry.is_process_alive(record):
+        async with qmp_session(record) as client:
+            await execute(client, "quit")
+        deadline = asyncio.get_event_loop().time() + 10
+        while pid_alive(record.pid) and asyncio.get_event_loop().time() < deadline:
+            await asyncio.sleep(0.1)
+
+    state.registry.remove(name)
+    shutil.rmtree(state.config.vm_state_dir(name), ignore_errors=True)
+    shutil.rmtree(state.config.vm_runtime_dir(name), ignore_errors=True)
+    return {
+        "name": name,
+        "destroyed": True,
+        "overlay_deleted": not Path(overlay).exists(),
+        "base_image_untouched": record.config.get("sandbox_base"),
+    }
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (sandbox_vm, sandbox_destroy):
+        mcp.tool(fn)

tests/test_sandbox.py

@@ -0,0 +1,145 @@
+"""Port-forward resolution and one-shot sandbox tools."""
+
+import socket
+
+import pytest
+from fastmcp import Client
+from fastmcp.exceptions import ToolError
+
+from conftest import result_data, write_registry
+from mcqemu.launcher import resolve_port_forwards
+from mcqemu.server import mcp
+from test_lifecycle import seeded_record
+
+
+def test_resolve_explicit_free_port():
+    assert resolve_port_forwards(["18765:22"]) == ["18765:22"]
+
+
+def test_resolve_auto_picks_free_port():
+    (entry,) = resolve_port_forwards(["auto:22"])
+    host, guest = entry.split(":")
+    assert guest == "22"
+    assert 1 <= int(host) <= 65535
+
+
+def test_resolve_bare_guest_port_means_auto():
+    (entry,) = resolve_port_forwards(["80"])
+    assert entry.endswith(":80")
+
+
+def test_resolve_busy_port_fails_fast():
+    with socket.socket() as blocker:
+        blocker.bind(("", 0))
+        busy = blocker.getsockname()[1]
+        with pytest.raises(ToolError, match="already in use"):
+            resolve_port_forwards([f"{busy}:22"])
+
+
+def test_resolve_rejects_garbage():
+    for bad in ["22->2222", "x:y", "auto:notaport", "0:0"]:
+        with pytest.raises(ToolError):
+            resolve_port_forwards([bad])
+
+
+@pytest.fixture
+def base_image(tmp_path):
+    """A real minimal qcow2 to serve as sandbox base."""
+    import subprocess
+
+    path = tmp_path / "base.qcow2"
+    subprocess.run(
+        ["qemu-img", "create", "-f", "qcow2", str(path), "256M"],
+        check=True,
+        capture_output=True,
+    )
+    return path
+
+
+@pytest.fixture
+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):
+        from mcqemu.launcher import resolve_port_forwards
+        from mcqemu.tools._common import app
+
+        state = app(ctx)
+        record = seeded_record(
+            dirs, name, accel="kvm", config={"disks": disks, "port_forwards": port_forwards}
+        )
+        state.registry.add(record)
+        return {
+            "name": name,
+            "status": "running",
+            "accel": "kvm",
+            "port_forwards": resolve_port_forwards(port_forwards),
+        }
+
+    monkeypatch.setattr("mcqemu.tools.sandbox.launch_vm", fake_launch_vm)
+
+
+@pytest.fixture
+def agent_up(monkeypatch):
+    async def responding(record):
+        return True
+
+    monkeypatch.setattr("mcqemu.tools.sandbox._agent_responding", responding)
+
+
+async def test_sandbox_vm_creates_overlay_and_waits(dirs, base_image, fake_launch, agent_up):
+    async with Client(mcp) as client:
+        data = result_data(await client.call_tool("sandbox_vm", {"base_image": str(base_image)}))
+    assert data["name"] == "sandbox"
+    assert data["sandbox"] is True
+    assert data["guest_agent"] == "responding"
+    assert data["port_forwards"][0].endswith(":22")
+    overlay = dirs.state / "vms" / "sandbox" / "overlay.qcow2"
+    assert overlay.exists()
+    assert data["overlay"] == str(overlay)
+
+
+async def test_sandbox_auto_names_avoid_collisions(dirs, base_image, fake_launch, agent_up):
+    write_registry(dirs, seeded_record(dirs, "sandbox"))
+    async with Client(mcp) as client:
+        data = result_data(await client.call_tool("sandbox_vm", {"base_image": str(base_image)}))
+    assert data["name"] == "sandbox-2"
+
+
+async def test_sandbox_agent_timeout_reports_hint(dirs, base_image, fake_launch, monkeypatch):
+    async def never(record):
+        return False
+
+    monkeypatch.setattr("mcqemu.tools.sandbox._agent_responding", never)
+    async with Client(mcp) as client:
+        data = result_data(
+            await client.call_tool("sandbox_vm", {"base_image": str(base_image), "wait_agent_s": 0})
+        )
+    assert data["guest_agent"] == "unavailable"
+    assert "qemu-guest-agent" in data["agent_hint"]
+
+
+async def test_sandbox_destroy_removes_overlay(dirs, all_pids_dead):
+    overlay = dirs.state / "vms" / "sb" / "overlay.qcow2"
+    overlay.parent.mkdir(parents=True)
+    overlay.write_bytes(b"fake")
+    write_registry(
+        dirs,
+        seeded_record(
+            dirs, "sb", config={"disks": [str(overlay)], "sandbox_overlay": str(overlay)}
+        ),
+    )
+    async with Client(mcp) as client:
+        data = result_data(await client.call_tool("sandbox_destroy", {"name": "sb"}))
+        assert data["destroyed"] is True
+        assert data["overlay_deleted"] is True
+        assert not overlay.exists()
+        vms = result_data(await client.call_tool("list_vms", {}))
+        assert vms == []
+
+
+async def test_sandbox_destroy_refuses_non_sandbox(dirs, all_pids_dead):
+    write_registry(dirs, seeded_record(dirs, "vm1"))
+    async with Client(mcp) as client:
+        with pytest.raises(ToolError, match="not created by sandbox_vm"):
+            await client.call_tool("sandbox_destroy", {"name": "vm1"})