#!/usr/bin/env python3 """Searching and opening a page must cost two calls, and stay outside the VM. python3 tests/test_vm_web_tools.py WHY Reading one web page in GUI mode used to cost six round trips through a 27B model with thinking on: screenshot to find the browser, click to focus it, click the address bar, type, Return, screenshot. Five of those six are operating a text field. So there are three tools instead: `web_search` and `read_page`, which run on the HOST against the same SearXNG and the same fetcher the main model uses, and `open_page`, which puts a URL on the screen in one call. What these cases protect: * the search must never be reachable FROM the machine. SearXNG sits in another container's network namespace behind a default-deny filter, and the whole sandbox argument rests on the guest not being able to reach the house's own services. * `open_page` is GUI-only. In CLI mode there is no desktop, and a tool that cannot work is a tool the model will spend a turn discovering that about. * the URL is passed to the guest as an environment variable, never interpolated into the shell script it runs. """ import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "vmagent", "app")) sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "vm", "vmd")) import browser BROWSER_SRC = open(browser.__file__, encoding="utf-8").read() # noqa: E402 import tools # noqa: E402 import websearch # noqa: E402 FAILS: list[str] = [] def check(label: str, condition: bool, detail: str = "") -> None: if condition: print(f"ok {label}") else: FAILS.append(f"{label}: {detail}") print(f"FAIL {label}\n {detail}") # --- what is offered, and where -------------------------------------- # ONE KIND OF SESSION. It used to be cli or gui, chosen by the calling # model, with the screen tools in one and the lookup suite in the other. # The caller chose wrong, a screenless session could build a window and # never see it, and a screen-only session reasoned about syntax instead of # looking it up. Every session now has all three. offered = [t["function"]["name"] for t in tools.schemas("ubuntu")] cli = gui = offered planning = [t["function"]["name"] for t in tools.schemas("ubuntu", booting=True)] for name in ("web_search", "read_page"): check(f"{name} is offered in CLI mode", name in cli, str(cli)) check(f"{name} is offered in GUI mode", name in gui, str(gui)) check("open_page is there, because every session has a screen", "open_page" in offered) check("none of them are offered while the machine boots", not {"web_search", "read_page", "open_page"} & set(planning), str(planning)) # --- the URL never becomes shell --------------------------------------- for bad in ("file:///etc/passwd", "javascript:alert(1)", "ftp://example.com", "http://user:pw@example.com/", "https://example.com/ ; rm -rf /", "https://example.com/`id`", "not a url", "", "https://" + "a" * 3000): try: browser.validate(bad) check(f"refuses {bad[:40]!r}", False, "it was accepted") except browser.BrowserError: check(f"refuses {bad[:40]!r}", True) # `$`, `(` and `)` are RFC 3986 sub-delims and real URLs are full of them - # every Wikipedia disambiguation page, for one. Rejecting them would break # ordinary use to defend against an injection that cannot happen: the URL # never enters a shell string. It is asserted inert below instead. for good in ("https://example.com", "http://example.com/a/b?c=d&e=f#g", "https://docs.python.org/3/library/os.html", "https://en.wikipedia.org/wiki/Python_(programming_language)", "https://example.com/$(id)"): try: browser.validate(good) check(f"accepts {good}", True) except browser.BrowserError as exc: check(f"accepts {good}", False, str(exc)) # THE URL IS SUBSTITUTED, AND IT USED TO ARRIVE IN THE ENVIRONMENT. # # Passing `env` to guest-exec REPLACES the child environment rather than # adding to it, so PATH disappears - a bare "powershell" could not be # resolved, and inside the script neither could `schtasks`. Measured: every # launch died with "Failed to execute child process (No such file or # directory)", including notepad.exe. `run` never showed it because # `shell()` passes no env at all, so this was latent in open_page on # Windows from the day it was written. # # So the value is substituted into the script instead, and the safety # question moves with it: a URL must not be able to become code. for script in (browser.LINUX_SCRIPT, browser.WINDOWS_SCRIPT): check("the script has exactly one slot for the URL and nothing else formatted", script.count("__URL__") == 1 and "{url}" not in script and "%s" not in script, script[:200]) check("no launcher passes an environment any more", "env=[" not in BROWSER_SRC, "that is what removed PATH") check("the value is quoted into a single-quoted literal", browser._ps_quote(chr(39)) == chr(39) * 2, "doubling the quote is what stops it closing the literal early") check("so a quote in a URL cannot end the string early", chr(39) not in browser._ps_quote("x" + chr(39) + "y").replace(chr(39) * 2, "")) check("the only other substitution is a constant", browser.CHROMIUM_FLAGS.startswith("--") and "__FLAGS__" in browser.LINUX_SCRIPT) check("and starts the browser as the desktop user, not as root", 'runuser -u "$user"' in browser.LINUX_SCRIPT) # systemd-run --user, not a hand-built environment: gnome-shell is the # compositor and never had WAYLAND_DISPLAY to hand down, so reading its # /proc/environ produced an X11 fallback that died. The user manager is # where a desktop exports what a launcher needs. check("through the user's own systemd instance, which has the display environment", "systemd-run --user" in browser.LINUX_SCRIPT) # Assert against the CODE, not the prose. The comments in that script # name both `pgrep` and the mistakes that are no longer in it, and a # naive `in` test reads a warning as the thing it warns about - the same # shape of error as `pgrep -f chromium` matching its own command line. code = "\n".join(line for line in browser.LINUX_SCRIPT.splitlines() if not line.lstrip().startswith("#")) check("and detaches by construction, so the call does not wait for a browser to close", "--unit=" in code and not any(l.rstrip().endswith("&") for l in code.splitlines()), "something is still backgrounded with &") check("and the launch is verified by that unit's state, not by a process name", "is-active" in code and "pgrep" not in code, "a pgrep survived in the code") # LocalSystem cannot draw on the logged-on user's desktop; a GUI process # started from session 0 succeeds and is invisible, which is worse than # failing. The scheduled task is what crosses into the session. check("windows crosses into the interactive session", "/it" in browser.WINDOWS_SCRIPT and "schtasks" in browser.WINDOWS_SCRIPT) check("and cleans the task up again", "schtasks /delete" in browser.WINDOWS_SCRIPT) # --- the search goes to the host's gateway, with a bearer token -------- sent: dict = {} def fake_post(path, payload): sent["path"], sent["payload"] = path, payload return "results" websearch._post = fake_post # noqa: SLF001 websearch.search("qemu guest agent", max_results=99, site="qemu.org") check("search posts to /web_search", sent["path"] == "/web_search", str(sent)) check("and clamps max_results to what the endpoint accepts", sent["payload"]["max_results"] == 15, str(sent["payload"])) check("and passes a site filter through", sent["payload"].get("site") == "qemu.org", str(sent["payload"])) websearch.read("https://example.com", max_chars=999999) check("read_page posts to /fetch_page", sent["path"] == "/fetch_page", str(sent)) check("and clamps the window", sent["payload"]["max_chars"] == 20000, str(sent["payload"])) for empty in ("", " "): try: websearch.search(empty) check("an empty search is refused", False, "it was accepted") except websearch.SearchError: check("an empty search is refused", True) break # --- and it is the HOST that searches, never the guest ----------------- # The verb list is the contract: if a future change moved the search into # the machine, it would have to pass through a guest command, and this is # where that would show up. vmd_source = open(os.path.join(os.path.dirname(__file__), "..", "vm", "vmd", "vmd.py")).read() start = vmd_source.index("def v_web_search") end = vmd_source.index("def v_open_page") check("web_search and read_page never touch the guest", "guest(" not in vmd_source[start:end] and "shell(" not in vmd_source[start:end], "a guest call appeared in a host-side verb") check("open_page does require a ready machine", "_require_ready()" in vmd_source[vmd_source.index("def v_open_page"): vmd_source.index("def v_reboot_vm")]) # --- the lookup suite, and whether it still matches the endpoints ------ # These schemas are hand-written so they stay terse - the agent pays for # every one of them on every turn - which means they can drift from the # endpoints they call. Measured the first time this ran: read_man_page # takes `name`, and the schema said `page`, so the agent's call came back # "Field required" from a tool that was working perfectly. # EVERY SESSION GETS THE LOOKUP SUITE, GUI INCLUDED. # # It used to be CLI-only, on the reasoning that a desktop session could # read anything it needed by opening it in the browser. True, and far too # expensive: open_page costs a screenshot per scroll where read_page costs # one call. Worse, a session that cannot cheaply look something up reasons # about it instead - measured, one spent eleven minutes and eight context # compactions deriving how to run background work in PowerShell, which the # documentation states outright, while a parse-clean 19 KB application it # had already written sat untouched. cli_names, gui_names = set(cli), set(gui) # Read back off the RENDERED schemas rather than a module-level list: the # definitions live in configs/agent/tools.toml now, so what matters is # what comes out of schemas(), not what went into it. research_schemas = [t for t in tools.schemas("ubuntu") if t["function"]["name"] in tools.RESEARCH_NAMES] for t in research_schemas: name = t["function"]["name"] check(f"{name} is offered in both modes", name in cli_names and name in gui_names, "being stuck costs more than a few extra tool schemas") check("and the screen tools are there alongside them", {"open_page", "screenshot", "click", "move", "key", "type", "scroll", "launch_app"} <= set(offered), str(sorted(offered))) try: import json as _json import urllib.request as _req spec = _json.load(_req.urlopen("http://172.22.0.10:8000/openapi.json", timeout=10)) except Exception as exc: # noqa: BLE001 print(f"skip the gateway is not reachable from here ({type(exc).__name__}); " "schema drift not checked") else: comps = spec["components"]["schemas"] for t in research_schemas: name = t["function"]["name"] try: body = spec["paths"][f"/tools/{name}"]["post"]["requestBody"] ref = body["content"]["application/json"]["schema"]["$ref"].rsplit("/", 1)[-1] real = comps[ref] except KeyError: check(f"{name} still exists on the gateway", False, "the endpoint is gone") continue ours = set(t["function"]["parameters"]["properties"]) theirs = set(real.get("properties", {})) invented = ours - theirs missing = set(real.get("required", [])) - set(t["function"]["parameters"].get("required", [])) check(f"{name}'s schema matches the endpoint", not invented and not missing, f"invented={sorted(invented)} missing-required={sorted(missing)}") # And the allowlist on vmd's side is the real gate, so it must name # tools that exist - and must not name the ones deliberately left out. sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "vm", "vmd")) import websearch as ws # noqa: E402 for name in ws.RESEARCH_TOOLS: check(f"the allowlist entry {name} is a real endpoint", f"/tools/{name}" in spec["paths"]) for forbidden in ("vm_agent", "matlab_run", "dns_lookup", "http_headers", "inspect_tls", "lookup_ip_or_asn"): check(f"{forbidden} is not reachable from a session", forbidden not in ws.RESEARCH_TOOLS, "the machine can answer that itself, or should never call it") print() if FAILS: print(f"FAILED {len(FAILS)}") raise SystemExit(1) print("the web tools cost two calls and stay outside the machine")