qemu and build: review fixes across the rig driver, boot script, and fetch helpers

Bounded waits and validated arguments in run.sh and ui-drive.sh, a seeded
settings directory and root-only staged rootfs permissions with their own
tests, qmp.py and imgtools.py hardening, the fetch scripts checking what they
download, and ASCII typography throughout. Each fix carries its test under
qemu/tests or tests/.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N3G6m9Aw5RyVY4ZowtKzEj

qemu/tests/test_qmp_drive.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 """Offline tests for qmp.py's drive(): the QMP socket and the control channel
-are faked, so this runs in well under two seconds with no VM.
+are faked, so this needs no VM and runs in a few seconds -- most of that is
+test_mismatches_are_fails_not_stops deliberately waiting out two real
+one-second timeouts to prove wait_json/wait_hit honour them.
 
 What is worth pinning is the contract the docstring makes: every step gets a
 results.jsonl row of ok / fail / fatal and the run CONTINUES, so one run
@@ -14,8 +16,10 @@ was never compared with the one the reference was captured under.
 import json
 import os
 import socket
+import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 
@@ -232,6 +236,23 @@ class PureHelpers(unittest.TestCase):
         self.assertTrue(qmp.apply_op("lt", 1, "2"))
         self.assertTrue(qmp.apply_op("eq", "connected", "connected"))
 
+    def test_apply_op_rejects_bad_combinations(self):
+        # eval_json and verb_assert_stat both catch (TypeError, ValueError)
+        # specifically so a malformed OP in a hand-written or generated
+        # script reads as a `fail` row with a reason, not a driver crash --
+        # that contract depends on apply_op actually raising these, which
+        # nothing exercised directly before.
+        with self.assertRaises(ValueError):
+            qmp.apply_op("bogus", 1, "1")
+        with self.assertRaises(TypeError):
+            qmp.apply_op("contains", 5, "1")
+
+    def test_flag_reads_an_optional_argv_pair_or_the_default(self):
+        argv = ["qmp.py", "sock", "drive", "s.txt", "out", "--size", "480"]
+        self.assertEqual(qmp.flag(argv, "--size"), "480")
+        self.assertEqual(qmp.flag(argv, "--ctl"), None)
+        self.assertEqual(qmp.flag(argv, "--ctl", "default"), "default")
+
     def test_parse_stats(self):
         got = qmp.parse_stats(FakeCtl("x").send("stats"))
         self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2, "idle": 0.0})
@@ -380,6 +401,102 @@ class DriveVerbs(unittest.TestCase):
         for row in rows[:-1]:
             self.assertIn("QMP socket closed", row["detail"])
 
+    def test_malformed_numeric_argument_is_fatal_for_the_step_not_a_crash(self):
+        # Several verbs parse their own arguments with bare
+        # int()/float()/positional indexing before any handler-local guard
+        # (tap, swipe, fling, sleep, wait_hit, wait_json, capture_region,
+        # wait_region). A typo'd coordinate or a missing argument -- exactly
+        # what a hand-edited *.txt script or a flowc.py bug can produce --
+        # used to raise ValueError/IndexError straight out of drive(),
+        # losing every row from that line onward instead of reading as its
+        # own fatal row (flare-edge #244).
+        rc, by, rows = run_script(
+            "tap 10 abc\n"
+            "sleep 0\n",
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
+        self.assertIn("invalid literal", by["tap 10 abc"]["detail"])
+
+        rc, by, rows = run_script(
+            "tap 10\n"
+            "sleep 0\n",
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
+        self.assertIn("list index out of range", by["tap 10"]["detail"])
+
+        rc, by, rows = run_script(
+            "capture_region r1 0 0 8 notanumber exact\n"
+            "sleep 0\n",
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
+        self.assertIn("invalid literal",
+                      by["capture_region r1 0 0 8 notanumber exact"]["detail"])
+
+    def test_bad_op_reads_as_a_fail_row_not_a_crash(self):
+        # apply_op's error paths (unknown OP -> ValueError, 'contains'
+        # against the wrong type -> TypeError) are caught by both callers
+        # (eval_json, verb_assert_stat) and must read as an ordinary `fail`
+        # row through the real verb handlers, not an uncaught exception or a
+        # SystemExit out of drive() itself.
+        rc, by, rows = run_script(
+            "assert_json a.b bogus 1\n"
+            "assert_stat fps bogus 1\n"
+        )
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fail", "fail"])
+        self.assertIn("bogus", by["assert_json a.b bogus 1"]["detail"])
+        self.assertIn("bogus", by["assert_stat fps bogus 1"]["detail"])
+
+    def test_assert_ocr_reports_no_tesseract_ocr_failure_and_match_or_not(self):
+        # assert_ocr's own surface -- the tesseract-not-installed fatal, the
+        # subprocess call, its exception net, and the final regex decision
+        # -- had no coverage at all: a regression here would only be caught
+        # by a live rig run against real tesseract. shutil.which and
+        # subprocess.run are swapped the same way rs485_send is above, since
+        # both are stdlib calls qmp.py makes directly, not seams of its own.
+        refs = {"r1": {"x": 0, "y": 0, "w": 8, "h": 8, "tolerance": "exact",
+                       "phash": "0" * 16, "structural": "00" * 32}}
+        saved_which, saved_run = qmp.shutil.which, qmp.subprocess.run
+
+        def restore():
+            qmp.shutil.which, qmp.subprocess.run = saved_which, saved_run
+
+        try:
+            qmp.shutil.which = lambda name: None
+            rc, by, rows = run_script("assert_ocr r1 hello\n", refs=refs)
+            self.assertEqual(rows[0]["status"], "fatal")
+            self.assertIn("tesseract not installed", rows[0]["detail"])
+
+            qmp.shutil.which = lambda name: "/usr/bin/tesseract"
+
+            def crashing_run(*a, **k):
+                raise OSError("tesseract crashed")
+
+            qmp.subprocess.run = crashing_run
+            rc, by, rows = run_script("assert_ocr r1 hello\n", refs=refs)
+            self.assertEqual(rows[0]["status"], "fatal")
+            self.assertIn("ocr failed", rows[0]["detail"])
+
+            def matching_run(cmd, **k):
+                return type("R", (), {"stdout": "hello world\n"})()
+
+            qmp.subprocess.run = matching_run
+            rc, by, rows = run_script("assert_ocr r1 hello\n", refs=refs)
+            self.assertEqual(rows[0]["status"], "ok")
+
+            def nonmatching_run(cmd, **k):
+                return type("R", (), {"stdout": "goodbye\n"})()
+
+            qmp.subprocess.run = nonmatching_run
+            rc, by, rows = run_script("assert_ocr r1 hello\n", refs=refs)
+            self.assertEqual(rows[0]["status"], "fail")
+            self.assertIn("goodbye", rows[0]["detail"])
+        finally:
+            restore()
+
     def test_region_faults_are_per_step_fatal(self):
         # A pre-seeded reference whose box does not fit a 64x64 screendump,
         # a name with no reference, a tolerance other than the captured one,
@@ -469,6 +586,38 @@ class DriveVerbs(unittest.TestCase):
         self.assertEqual(rows[0]["status"], "fatal")
         self.assertIn("must not contain '/'", rows[0]["detail"])
 
+    def test_wait_region_hits_the_same_fatal_paths_as_assert_region(self):
+        # wait_region drives the same fresh_region() call as assert_region
+        # (comment on verb_wait_region), so a missing reference, a foreign
+        # tolerance, and a refs.json name that could escape outdir must all
+        # be fatal here too -- and, since none of them can ever start
+        # passing, each must stop on its first check instead of waiting out
+        # TIMEOUT_S (poll_until's check_region() signals this by returning
+        # ok=True, caught via the `fault` list).
+        refs = {
+            "r1": {"x": 0, "y": 0, "w": 8, "h": 8, "tolerance": "exact",
+                   "phash": "0" * 16, "structural": "00" * 32},
+            "../evil": {"x": 0, "y": 0, "w": 8, "h": 8, "tolerance": "exact",
+                        "phash": "0" * 16, "structural": "00" * 32},
+        }
+        t0 = time.monotonic()
+        rc, by, rows = run_script(
+            "wait_region nope exact 2\n"
+            "wait_region r1 loose 2\n"
+            "wait_region ../evil exact 2\n",
+            refs=refs,
+        )
+        waited = time.monotonic() - t0
+        self.assertEqual(rc, 1)
+        self.assertEqual([r["status"] for r in rows], ["fatal", "fatal", "fatal"])
+        self.assertIn("no reference", by["wait_region nope exact 2"]["detail"])
+        self.assertIn("captured as exact, script expects loose",
+                      by["wait_region r1 loose 2"]["detail"])
+        self.assertIn("must not contain '/'", by["wait_region ../evil exact 2"]["detail"])
+        self.assertLess(waited, 2.0,
+                        "none of these three can ever pass, so none may wait "
+                        "out its TIMEOUT_S of 2s each")
+
     def test_unknown_verb_is_fatal_for_the_run(self):
         # A silently-ignored line is a test that proves nothing, so this one
         # is the documented exception to "the run continues": drive() exits
@@ -576,5 +725,251 @@ class FullscreenToggleTracksRealState(unittest.TestCase):
                           "the already-off real state instead of turning it on")
 
 
+class LazyImports(unittest.TestCase):
+    def test_module_import_does_not_pull_in_imgtools_or_pillow(self):
+        # imgtools.py imports Pillow at its own module scope; qmp.py used to
+        # `import imgtools` at ITS module scope too, so every subprocess
+        # invocation of qmp.py paid that cost even for screendump/tap/quit,
+        # which never touch a pixel -- defeating the Pillow-free boot-wait
+        # loop ui-drive.sh's own comment documents. A subprocess (not just
+        # checking qmp.imgtools in-process) is what actually pins this: the
+        # other tests in this file exercise region verbs and so leave the
+        # lazy slot filled in for the rest of THIS process.
+        script = (
+            "import sys\n"
+            f"sys.path.insert(0, {HERE!r})\n"
+            "import qmp\n"
+            "assert 'imgtools' not in sys.modules, 'imgtools imported eagerly'\n"
+            "assert 'PIL' not in sys.modules, 'Pillow imported eagerly'\n"
+        )
+        result = subprocess.run([sys.executable, "-c", script],
+                                capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+
+def _bare_ctl(sock):
+    """A Ctl instance around an already-connected socket, bypassing
+    __init__'s own socket()+connect() (there is no path on disk to connect
+    to -- these tests drive a socketpair() end directly)."""
+    ctl = qmp.Ctl.__new__(qmp.Ctl)
+    ctl.sock = sock
+    ctl.buf = b""
+    return ctl
+
+
+class CtlSocketProtocol(unittest.TestCase):
+    """Ctl.send() itself -- the line-buffering loop that reassembles a reply
+    across possibly many recv() calls, skips the cooked-mode echo of the
+    command it just sent, and stops on the SENTINEL line -- has zero
+    coverage anywhere else in this file: every FakeCtl/DyingCtl/etc. above
+    replaces the whole class, never exercising the real one. This drives the
+    real qmp.Ctl over a live AF_UNIX socketpair standing in for the FIFO
+    bridge in rootfs/sbin/init, so the actual wire protocol gets checked
+    without a VM or rootfs changes."""
+
+    def setUp(self):
+        self.client_sock, self.server_sock = socket.socketpair(
+            socket.AF_UNIX, socket.SOCK_STREAM)
+        self.client_sock.settimeout(5.0)
+        self.ctl = _bare_ctl(self.client_sock)
+
+    def tearDown(self):
+        self.client_sock.close()
+        self.server_sock.close()
+
+    def test_reassembles_a_reply_split_across_two_recv_calls(self):
+        # The reply plus SENTINEL arrive in two separate writes, forcing
+        # Ctl.send() through at least two recv() calls for one line: the
+        # exact shape a reply straddling a 4096-byte read boundary takes on
+        # real hardware.
+        def server():
+            self.server_sock.recv(4096)  # the command line
+            self.server_sock.sendall(b"first line\nsecond ")
+            time.sleep(0.05)
+            self.server_sock.sendall(b"line\n<<END>>\n")
+
+        th = threading.Thread(target=server, daemon=True)
+        th.start()
+        try:
+            got = self.ctl.send("stats")
+        finally:
+            th.join(timeout=2)
+        self.assertEqual(got, "first line\nsecond line")
+
+    def test_strips_the_cooked_mode_echo_of_the_command(self):
+        # The tty is in cooked mode, so the command comes back echoed before
+        # the real reply; Ctl.send() must drop that line, not treat it as
+        # part of the answer.
+        def server():
+            cmd_line = self.server_sock.recv(4096)
+            self.server_sock.sendall(cmd_line)  # cooked-mode echo
+            self.server_sock.sendall(b"the actual reply\n<<END>>\n")
+
+        th = threading.Thread(target=server, daemon=True)
+        th.start()
+        try:
+            got = self.ctl.send("page")
+        finally:
+            th.join(timeout=2)
+        self.assertEqual(got, "the actual reply")
+
+    def test_leftover_bytes_after_sentinel_carry_over_to_the_next_send(self):
+        # One write carries this reply's SENTINEL immediately followed by
+        # bytes belonging to the NEXT command's reply -- proving self.buf
+        # correctly holds the leftover across two separate send() calls
+        # instead of dropping or re-reading it.
+        def server():
+            self.server_sock.recv(4096)
+            self.server_sock.sendall(b"reply one\n<<END>>\nreply two\n<<END>>\n")
+
+        th = threading.Thread(target=server, daemon=True)
+        th.start()
+        try:
+            got1 = self.ctl.send("cmd1")
+        finally:
+            th.join(timeout=2)
+        self.assertEqual(got1, "reply one")
+
+        # cmd2's own reply is already sitting in self.ctl.buf from the single
+        # write above; send() must serve it without another recv().
+        got2 = self.ctl.send("cmd2")
+        self.assertEqual(got2, "reply two")
+
+
+class CtlBufferCap(unittest.TestCase):
+    """Regression for Ctl.send() growing self.buf without bound: a peer that
+    keeps streaming bytes fast enough to beat the per-recv() socket timeout,
+    but never emits a newline or SENTINEL, used to grow self.buf forever
+    instead of failing closed."""
+
+    def test_raises_instead_of_growing_self_buf_without_bound(self):
+        client_sock, server_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
+        client_sock.settimeout(3.0)
+        ctl = _bare_ctl(client_sock)
+
+        def server():
+            server_sock.recv(4096)
+            target = qmp.Ctl.MAX_BUF + 8192
+            sent = 0
+            try:
+                while sent < target:
+                    server_sock.sendall(b"x" * 4096)  # no newline, ever
+                    sent += 4096
+            except OSError:
+                pass  # the client closed once the cap tripped; nothing left to send to
+
+        th = threading.Thread(target=server, daemon=True)
+        th.start()
+        try:
+            with self.assertRaises(RuntimeError) as cm:
+                ctl.send("stats")
+            self.assertIn("too large", str(cm.exception))
+            self.assertLessEqual(
+                len(ctl.buf), qmp.Ctl.MAX_BUF + 4096,
+                "must fail as soon as the cap is crossed, not keep draining "
+                "an unbounded peer first")
+        finally:
+            client_sock.close()
+            server_sock.close()
+            th.join(timeout=2)
+
+
+class QmpSocketTimeout(unittest.TestCase):
+    """Regression for the QMP unix socket having no timeout: a peer that
+    accepts the connection but never answers (a wedged VM -- a TCG stall or
+    a kernel panic loop) used to block main()'s greeting readline() forever.
+    ui-drive.sh's own cleanup() calls `quit` on this exact socket before it
+    reaches reap("$QEMU_PID"), so an unbounded hang here defeats the one
+    thing meant to guarantee a wedged qemu-system-arm cannot outlive the
+    script."""
+
+    def test_main_bounds_a_wedged_qmp_peer_instead_of_hanging_forever(self):
+        d = tempfile.mkdtemp(prefix="qmpsock.")
+        sock_path = os.path.join(d, "qmp.sock")
+        srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        srv.bind(sock_path)
+        srv.listen(1)
+
+        def accept_and_hang():
+            conn, _ = srv.accept()
+            time.sleep(5)  # never answer the greeting/qmp_capabilities handshake
+            conn.close()
+
+        th = threading.Thread(target=accept_and_hang, daemon=True)
+        th.start()
+        saved_timeout, saved_argv = qmp.QMP_TIMEOUT_S, sys.argv
+        qmp.QMP_TIMEOUT_S = 0.3
+        sys.argv = ["qmp.py", sock_path, "quit"]
+        try:
+            t0 = time.monotonic()
+            with self.assertRaises(OSError):
+                qmp.main()
+            elapsed = time.monotonic() - t0
+            self.assertLess(
+                elapsed, 2.0,
+                "a wedged QMP peer must be bounded by QMP_TIMEOUT_S, not hang "
+                "indefinitely (main()'s socket needs its own settimeout(), the "
+                "same way Ctl's already has one)")
+        finally:
+            sys.argv = saved_argv
+            qmp.QMP_TIMEOUT_S = saved_timeout
+            srv.close()
+            th.join(timeout=6)
+
+
+class JsonStatusFetch(unittest.TestCase):
+    """fetch_status_json()'s two failure branches -- the guest's snapshot not
+    existing yet (the bridge answers 'bridge: no such file: PATH' for the
+    first couple of seconds after boot, before webstatus.c's first 2s timer
+    tick) and a torn/invalid JSON snapshot -- have no coverage anywhere else
+    in this file: every FakeCtl-style '@cat' handler above always returns
+    valid JSON."""
+
+    class StubCtl:
+        def __init__(self, reply):
+            self.reply = reply
+
+        def send(self, cmd):
+            assert cmd.startswith("@cat "), cmd
+            return self.reply
+
+    def test_missing_snapshot_file_is_a_detail_not_a_crash(self):
+        ctl = self.StubCtl("bridge: no such file: /tmp/warden-web-status.json")
+        doc, err = qmp.fetch_status_json(ctl)
+        self.assertIsNone(doc)
+        self.assertEqual(err, "bridge: no such file: /tmp/warden-web-status.json")
+
+    def test_torn_json_is_a_detail_not_a_crash(self):
+        ctl = self.StubCtl('{"a": 1, "b":')
+        doc, err = qmp.fetch_status_json(ctl)
+        self.assertIsNone(doc)
+        self.assertIn("status json unparsable", err)
+
+    def test_eval_json_turns_a_missing_snapshot_into_an_ordinary_fail(self):
+        ctl = self.StubCtl("bridge: no such file: /tmp/warden-web-status.json")
+        ok, detail = qmp.eval_json(ctl, "a.b", "eq", "1")
+        self.assertFalse(ok)
+        self.assertEqual(detail, "bridge: no such file: /tmp/warden-web-status.json")
+
+    def test_eval_json_turns_torn_json_into_an_ordinary_fail(self):
+        ctl = self.StubCtl('{"a": 1, "b":')
+        ok, detail = qmp.eval_json(ctl, "a.b", "eq", "1")
+        self.assertFalse(ok)
+        self.assertIn("status json unparsable", detail)
+
+    def test_wait_json_retries_a_missing_snapshot_instead_of_treating_it_fatal(self):
+        # A missing snapshot is an ordinary not-yet-true check, so wait_json
+        # must poll it out to TIMEOUT_S like any other fail -- not read the
+        # bridge's plain-text error as a channel fault the way a dead
+        # RuntimeError/OSError from ctl.send() itself already is.
+        ctl = self.StubCtl("bridge: no such file: /tmp/warden-web-status.json")
+        t0 = time.monotonic()
+        ok, detail, waited, fatal = qmp.poll_until(
+            lambda: qmp.eval_json(ctl, "a.b", "eq", "1"), 0.6, period=0.2)
+        self.assertFalse(ok)
+        self.assertFalse(fatal, "a missing snapshot is a fail to retry, not a channel fault")
+        self.assertGreaterEqual(time.monotonic() - t0, 0.6)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=1)