diff --git a/qemu/rootfs/sbin/init b/qemu/rootfs/sbin/init index 1c072fd..1df3557 100755 --- a/qemu/rootfs/sbin/init +++ b/qemu/rootfs/sbin/init @@ -10,6 +10,17 @@ exec /dev/console 2>&1 /bin/busybox --install -s /bin +# mkfs.ext4 -d preserves the checkout owner's uid. Production owns shadow as +# root, and the UI intentionally rejects any other owner before verifying it. +chown 0:0 /etc/shadow +chmod 0600 /etc/shadow +root_hash="$(awk -F: '$1 == "root" { print $2 }' /etc/shadow)" +test_hash="$(printf '%s' root | /usr/bin/mkpasswd -m md5 -S wardenrs 2>/dev/null)" +if [ -z "$root_hash" ] || [ "$test_hash" != "$root_hash" ]; then + echo "FATAL: QEMU root credential verifier is unavailable" + poweroff -f +fi + mount -t proc proc /proc mount -t sysfs sysfs /sys mount -t tmpfs tmpfs /tmp diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index f71353c..6c6bce4 100755 --- a/qemu/tests/qmp.py +++ b/qemu/tests/qmp.py @@ -148,6 +148,8 @@ def _load_imgtools(): return imgtools AXIS_MAX = 32767 +TAP_HOLD_S = 0.35 +TAP_OBSERVE_TIMEOUT_S = 2.0 # Bounds every blocking read on the QMP socket -- the greeting banner, the # qmp_capabilities handshake, and every screendump/tap/quit round trip -- @@ -206,15 +208,37 @@ def to_axis(px, size): return min(AXIS_MAX, math.ceil(px * AXIS_MAX / (size - 1))) -def do_tap(s, f, ax, ay, hold=0.2): +def do_tap(s, f, ax, ay, hold=TAP_HOLD_S): send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)]) - # Hold the press across several LVGL indev poll periods (33 ms each): an - # instantaneous press+release lands inside one poll and no click is ever - # registered. + # Hold the press across several LVGL indev poll periods (33 ms each), but + # stay below LVGL's 400 ms long-press threshold. A longer hold repeats + # controls such as Backspace and no longer represents a tap. time.sleep(hold) send_events(s, f, [btn_ev(False)]) +def do_observed_tap(ctx, ax, ay): + """Press until the guest reports consuming it, then release.""" + before = parse_stats(ctx.ctl.send("stats")).get("presses") + if before is None: + do_tap(ctx.s, ctx.f, ax, ay) + return True + + send_events(ctx.s, ctx.f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)]) + observed = False + deadline = time.monotonic() + TAP_OBSERVE_TIMEOUT_S + try: + while time.monotonic() < deadline: + now = parse_stats(ctx.ctl.send("stats")).get("presses") + if now is not None and now != before: + observed = True + break + time.sleep(0.02) + finally: + send_events(ctx.s, ctx.f, [btn_ev(False)]) + return observed + + def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None): """Drag with interpolated motion. @@ -532,6 +556,11 @@ STATS_FIELD_RE = { "fps": re.compile(r'^fps:\s*(-?\d+(?:\.\d+)?)\s*$'), "render": re.compile(r'^render:\s*(-?\d+(?:\.\d+)?)\s*ms/frame\s*$'), "idle": re.compile(r'^idle:\s*(\d+)\s*$'), + "presses": re.compile(r'^presses:\s*(\d+)\s*$'), + "termbusy": re.compile(r'^termbusy:\s*(\d+)\s*$'), + "termintr": re.compile(r'^termintr:\s*(\d+)\s*$'), + "termfg": re.compile(r'^termfg:\s*(-?\d+)\s*$'), + "termsig": re.compile(r'^termsig:\s*(\d+)\s*$'), } @@ -689,8 +718,8 @@ def verb_shot(ctx, lineno, cmd, args, line): def verb_tap(ctx, lineno, cmd, args, line): x, y = int(args[0]), int(args[1]) - do_tap(ctx.s, ctx.f, to_axis(x, ctx.size), to_axis(y, ctx.size)) - return "ok", "" + observed = do_observed_tap(ctx, to_axis(x, ctx.size), to_axis(y, ctx.size)) + return ("ok", "") if observed else ("fail", "press was not consumed within 2 seconds") def verb_swipe(ctx, lineno, cmd, args, line): @@ -838,9 +867,9 @@ def verb_assert_stat(ctx, lineno, cmd, args, line): # assert_stat FIELD OP VALUE: FIELD off a fresh `stats` reply. ctx.need_ctl(lineno, cmd) field, op, value = args[0], args[1], args[2] - if field not in ("cpu", "fps", "render", "idle"): + if field not in ("cpu", "fps", "render", "idle", "presses", "termbusy", "termintr", "termfg", "termsig"): return "fail", (f"unknown stat field {field!r} " - f"(known: cpu, fps, render, idle)") + f"(known: cpu, fps, render, idle, presses, termbusy, termintr, termfg, termsig)") stats = parse_stats(ctx.ctl.send("stats")) if field not in stats: return "fail", "no such field" diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py index 10fc6c9..0ba9e30 100755 --- a/qemu/tests/test_qmp_drive.py +++ b/qemu/tests/test_qmp_drive.py @@ -22,6 +22,7 @@ import tempfile import threading import time import unittest +from types import SimpleNamespace HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) @@ -39,12 +40,16 @@ class FakeCtl: def __init__(self, path, timeout=15.0): self.path = path + self.stats_calls = 0 def send(self, cmd): if cmd == "page": return "Demo/Rows" if cmd == "stats": - return "page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\nrga: 0%\nidle: 0" + self.stats_calls += 1 + return ("page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\n" + f"rga: 0%\nidle: 0\npresses: {self.stats_calls}\ntermbusy: 1\n" + "termintr: 2\ntermfg: -1\ntermsig: 3") if cmd == "wake": return "wake: ok" if cmd == "home": @@ -212,6 +217,71 @@ def run_script(text, refs=None, rs485_control=None, ctl_cls=None, rpc_fn=None, c class PureHelpers(unittest.TestCase): + def test_tap_holds_press_across_loaded_guest_polls(self): + calls = [] + saved_send, saved_sleep = qmp.send_events, qmp.time.sleep + qmp.send_events = lambda _s, _f, events: calls.append(events) + qmp.time.sleep = lambda seconds: calls.append(seconds) + try: + qmp.do_tap(None, None, 123, 456) + finally: + qmp.send_events, qmp.time.sleep = saved_send, saved_sleep + + self.assertEqual(calls[0], [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)]) + self.assertGreaterEqual(calls[1], 0.25) + self.assertLess(calls[1], 0.4) + self.assertEqual(calls[2], [qmp.btn_ev(False)]) + + def test_observed_tap_releases_after_guest_consumes_press(self): + calls = [] + + class ObservingCtl: + def __init__(self): + self.calls = 0 + + def send(self, cmd): + self.calls += 1 + return f"presses: {0 if self.calls < 3 else 1}" + + saved_send, saved_sleep = qmp.send_events, qmp.time.sleep + qmp.send_events = lambda _s, _f, events: calls.append(events) + qmp.time.sleep = lambda _seconds: None + try: + observed = qmp.do_observed_tap( + SimpleNamespace(s=None, f=None, ctl=ObservingCtl()), 123, 456) + finally: + qmp.send_events, qmp.time.sleep = saved_send, saved_sleep + + self.assertTrue(observed) + self.assertEqual(calls, [ + [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)], + [qmp.btn_ev(False)], + ]) + + def test_observed_tap_timeout_still_releases_press(self): + calls = [] + clock = iter((0.0, 0.0, qmp.TAP_OBSERVE_TIMEOUT_S + 0.1)) + + class UnobservingCtl: + def send(self, cmd): + return "presses: 0" + + saved_send, saved_sleep, saved_monotonic = ( + qmp.send_events, qmp.time.sleep, qmp.time.monotonic) + qmp.send_events = lambda _s, _f, events: calls.append(events) + qmp.time.sleep = lambda _seconds: None + qmp.time.monotonic = lambda: next(clock) + try: + observed = qmp.do_observed_tap( + SimpleNamespace(s=None, f=None, ctl=UnobservingCtl()), 123, 456) + finally: + qmp.send_events = saved_send + qmp.time.sleep = saved_sleep + qmp.time.monotonic = saved_monotonic + + self.assertFalse(observed) + self.assertEqual(calls[-1], [qmp.btn_ev(False)]) + def test_every_pixel_round_trips_through_lvgl_calibration(self): # lv_evdev.c _evdev_calibrate: px = axis * (width - 1) / AXIS_MAX, # integer division. A tap requested at px must land at px, for every @@ -255,7 +325,10 @@ class PureHelpers(unittest.TestCase): 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}) + self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2, + "idle": 0.0, "presses": 1.0, + "termbusy": 1.0, "termintr": 2.0, + "termfg": -1.0, "termsig": 3.0}) def test_poll_until_turns_a_channel_fault_into_fatal_not_a_raise(self): # A CHECK that raises RuntimeError or OSError (Ctl.send on EOF or a