diff --git a/forgetest/forgetest/manifest.py b/forgetest/forgetest/manifest.py index 6954539..5a6a2c1 100644 --- a/forgetest/forgetest/manifest.py +++ b/forgetest/forgetest/manifest.py @@ -173,6 +173,32 @@ HOST_TESTED = [ # routes and a package's jog through the host's credential are # tests/host_routes.sh). ("forgectrl", "src/grblport.*", "forgectrl:tests/grblport_test.c", "forgectrl:.github/workflows/build.yml"), + # The panel page as the daemon serves it: bundled, the theme and the + # save bar in it, and its policy frame-src 'none'. + ("forgectrl", "src/ui.h", "forgectrl:tests/host_routes.sh", "forgectrl:.github/workflows/build.yml"), + # The data and run directories from the environment: the SSH flag in the + # run directory, written and removed by POST /system/ssh. + ("forgectrl", "src/paths.h", "forgectrl:tests/host_routes.sh", "forgectrl:.github/workflows/build.yml"), + # The first run in the page's order: the four documents at their hashes, + # the press step breathing teal, the press taken as the acceptance, the + # form steps through main.c's hooks (extensions.md stays with + # exthost.service, which covers it). + ("forgectrl", "src/advisories.*", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "docs/advisories/safety-and-risk.md", "forgectrl:tests/host_setup.sh", + "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "docs/advisories/licenses.md", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "docs/advisories/privacy.md", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "docs/advisories/cloud-service.md", "forgectrl:tests/host_setup.sh", + "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "src/button.*", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "src/led.*", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + ("forgectrl", "src/hooks.h", "forgectrl:tests/host_setup.sh", "forgectrl:.github/workflows/build.yml"), + # The setup sheet id from the fuse serial, against golden ids. + ("forgectrl", "src/sheetid.*", "forgectrl:tests/sheetid_test.c", "forgectrl:.github/workflows/build.yml"), + # The logging path: the image's rsyslog.conf and the rendered rules, each + # logger in its own file at its level, a stray tag in system/, the kern + # facility in kernel/. + ("forgectrl", "src/fflog.*", "forgectrl:tests/fflog_e2e.sh", "forgectrl:.github/workflows/build.yml"), ] # This repository's root, when the package runs from a checkout of it; a diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 0c6c161..e31d4cf 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -12,7 +12,6 @@ from . import forgectrl # noqa: F401,E402 from . import setup # noqa: F401,E402 from . import setup_dark # noqa: F401,E402 from . import setup_burns # noqa: F401,E402 -from . import logs # noqa: F401,E402 from . import motion # noqa: F401,E402 from . import cooling # noqa: F401,E402 from . import laser # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/camera.py b/forgetest/forgetest/suite/camera.py index 3bb97a3..22e1c73 100644 --- a/forgetest/forgetest/suite/camera.py +++ b/forgetest/forgetest/suite/camera.py @@ -55,7 +55,7 @@ def _jpeg_size(data): @test("camera.snapshot", title="Lid camera snapshot and stream", subsystem="camera", kind="auto", est_min=2, - covers=_CAM_COVERS, requires=["forgectrl.panel-serves"], + covers=_CAM_COVERS, steps=["Setup: lid closed."], description="/cam/snapshot returns a JPEG of a plausible size (a black frame compresses " "far smaller), a second snapshot with the lid lamp switched off differs from it " @@ -169,7 +169,7 @@ def snapshot(ctx): @test("camera.sensor-profile", title="Camera geometry follows the fitted sensor", subsystem="camera", kind="auto", est_min=1, - covers=_CAM_COVERS, requires=["forgectrl.panel-serves"], + covers=_CAM_COVERS, description="/cam/status names the sensor that bound (OV5648 on a 5 MP machine, OV8856 on an " "8 MP 'HD' one) and reports the geometry that sensor implies, and the JPEGs it " "actually returns are that size. A machine that reports 'unknown', or whose frames " @@ -212,7 +212,7 @@ def sensor_profile(ctx): @test("camera.h264-stream", title="H.264 live stream", subsystem="camera", kind="auto", est_min=2, - covers=_CAM_COVERS, requires=["forgectrl.panel-serves"], + covers=_CAM_COVERS, steps=["Setup: lid closed."], description="/cam/h264 answers a fragmented MP4: a codec header naming the SPS profile, an " "init segment (ftyp+moov) followed by media fragments (moof+mdat) within a few " @@ -274,7 +274,7 @@ def h264_stream(ctx): @test("camera.frame-health", title="Capture delivers whole frames", subsystem="camera", kind="auto", est_min=1, - covers=_CAM_COVERS, requires=["forgectrl.panel-serves"], + covers=_CAM_COVERS, description="The capture queue flags a frame errored when it is short, torn, or arrived " "after the CSI-2 receiver lost sync; forgectrl drops those rather than " "demosaicing them, restarts the stream if they persist, and withholds the " @@ -362,7 +362,7 @@ def frame_health(ctx): @test("camera.lid-privacy", title="Cameras see the room only with the lid closed", subsystem="camera", kind="operator", est_min=3, - covers=_PRIVACY_COVERS + [("forgectrl", "src/auth.*")], requires=["forgectrl.panel-serves"], + covers=_PRIVACY_COVERS + [("forgectrl", "src/auth.*")], actions=["lid"], steps=["Start with the lid closed; you will be told to open it, then close it again.", "Nothing moves and the laser is not involved."], diff --git a/forgetest/forgetest/suite/forgectrl.py b/forgetest/forgetest/suite/forgectrl.py index 93f2aab..b0d91c5 100644 --- a/forgetest/forgetest/suite/forgectrl.py +++ b/forgetest/forgetest/suite/forgectrl.py @@ -5,7 +5,6 @@ """forgectrl.* - the machine-services daemon's API, access control, and panel.""" import json -import os import socket import time @@ -312,142 +311,3 @@ def settings_bounds(ctx): ctx.check(applied == try_level, "lamp did not follow lid_lamp_idle=%s (reads %s)", try_level, applied) ctx.check(back == want, "lamp did not return to %s after the restore (reads %s)", want, back) ctx.log("lid lamp follows the setting live and returns to %s", want) - - -@test("forgectrl.panel-serves", title="Control panel and status endpoints", subsystem="forgectrl", - kind="auto", est_min=1, - covers=[("forgectrl", "src/ui.*"), ("forgectrl", "src/ui/**"), ("forgectrl", "src/status.*"), - ("forgectrl", "src/cam.c"), ("forgectrl", "src/main.c"), ("forgectrl", "src/super.c"), - ("grblhal-glowforge", "src/glowforge_status.c"), ("grblhal-glowforge", "src/serial.c"), - ("forgectrl", "src/curverec.*")], - description="The panel page is served, with a policy that lets no frame navigate anywhere " - "(frame-src 'none': a package's page renders in a frame of it, and a frame that " - "could go to another address could carry what it was shown in the URL; what the " - "header does in each browser is the frame-isolation harness's to prove, and this " - "holds the machine's own page to carrying it), /status carries the machine " - "telemetry the panel and " - "the acceptance tool read (including the sys block: CPU busy percent over the " - "interval since the previous read, memory used percent; and homed_axes, " - "the axes that carry a reference, Z alone once the lens has taken its " - "own at the controller's start), and /cam/status " - "answers. In GRBL mode with a live controller, /status also echoes the " - "controller's published state file as the grbl block (fresh age, machine " - "state, sender session, laser window and dose model, modal report), " - "GET /grbl/settings serves the published $$ view, and the controller's " - "settings store is /data/forgefirm/EEPROM-glowforge.DAT with nothing of " - "it at the top of /data.") -def panel_serves(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - st, body = fc.get("/", raw=True) - ev["panel_status"] = st - ctx.log("GET / -> %s (%d bytes)", st, len(body) if body else 0) - ctx.check(st == 200, "GET / -> %s", st) - text = body.decode("utf-8", "replace") - ctx.check(" %s %s", st, cam) - ctx.check(st == 200 and isinstance(cam, dict) and "running" in cam, "GET /cam/status -> %s", st) - - # The controller's published state, echoed only while a live GRBL - # controller runs (glowforge_status.c -> /run/forgefirm -> /status). - st, mode = fc.get("/mode") - if isinstance(mode, dict) and mode.get("mode") == "grbl" and mode.get("controller") == "running": - g = ctx.forgectrl.status().get("grbl") or {} - ev["grbl"] = g - ctx.log("/status grbl=%s", g) - ctx.check(isinstance(g.get("age_s"), (int, float)) and g["age_s"] < 30, - "/status grbl block missing or stale: %s", g) - rep = g.get("report") or {} - for key in ("state", "sender", "laser", "modals"): - ctx.check(key in rep, "/status grbl.report lacks %r", key) - ctx.check((rep.get("laser") or {}).get("model") in ("density", "analog"), - "grbl.report.laser carries no model: %s", rep.get("laser")) - ctx.check((rep.get("laser") or {}).get("curve"), - "grbl.report.laser carries no dose curve: %s", rep.get("laser")) - st, text = fc.get("/grbl/settings", raw=True) - ev["grbl_settings_status"] = st - ctx.check(st == 200 and b"$35=" in (text or b""), - "GET /grbl/settings -> %s without the $$ view", st) - # The $-settings persist in the data directory, never loose in /data. - nvs = "/data/forgefirm/EEPROM-glowforge.DAT" - ev["grbl_nvs"] = os.path.isfile(nvs) - ctx.check(os.path.isfile(nvs), "the controller's settings store is not at %s", nvs) - ctx.check(not os.path.exists("/data/EEPROM-glowforge.DAT"), - "a settings store remains at the top of /data") - else: - ctx.log("no live GRBL controller (%s); grbl block checks skipped", mode) - - # The dose-curve recorder's read surface answers in any mode. - st, cs = fc.get("/curve/status") - ev["curve_status"] = cs - ctx.check(st == 200 and isinstance(cs, dict) and cs.get("state") in - ("idle", "waiting", "recording", "done", "failed"), - "GET /curve/status -> %s %s", st, cs) - st, text = fc.get("/curve/ladder.gcode", raw=True) - ctx.check(st == 200 and b"S1000" in (text or b"") and b"M5" in (text or b""), - "GET /curve/ladder.gcode -> %s without the ladder", st) diff --git a/forgetest/forgetest/suite/logs.py b/forgetest/forgetest/suite/logs.py deleted file mode 100644 index 8a013a9..0000000 --- a/forgetest/forgetest/suite/logs.py +++ /dev/null @@ -1,187 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""logs.* - unified logging: the routing path (emitters, relays, rendered -rules); the level settings.""" -import os -import re -import time - -from ..catalog import test -from .. import hw - - -def _read(path, default=None): - try: - with open(path, "r", encoding="utf-8", errors="replace") as f: - return f.read() - except OSError: - return default - - -LOGS_ROOT = "/data/log/forgefirm" - - -# The routing test proves the whole path every logger takes: emitter (or -# relay) -> /dev/log -> rsyslog rules rendered from the settings -> the -# logger's own file in the ff_line format. It stands for the emitters in -# every component (fflog in forgectrl and grblHAL, the SysLogHandler in the -# Python apps), the supervisor's per-controller relay, the daemon's fifo -# relay in its init script, and the render step. -_ROUTING_COVERS = [("forgectrl", "src/logs.*"), ("forgectrl", "src/fflog.*"), ("forgectrl", "src/super.c"), - ("forgectrl", "init/**"), - ("grblhal-glowforge", "src/fflog.*"), ("grblhal-glowforge", "src/main.c"), - ("forgefirm-app", "forgefirm-app/ffmachine.py"), ("forgefirm-app", "forgefirm-app/gfcloud.py"), - ("forgefirm-app", "forgefirm-app/gfhome.py")] - -LOGGERS =("forgectrl", "grblhal", "gfcloud", "gfhome", "forgeext", "kernel", "system") -_LINE_RE = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(\.\d+)?[+-]\d\d:\d\d (?P[A-Za-z0-9_.-]+)\[(?P[-\d]+)\] " - r"(?PEMERG|ALERT|CRIT|ERR|WARNING|NOTICE|INFO|DEBUG) (?P.*)$") -_SEV_RANK = {"off": -1, "error": 3, "warning": 4, "notice": 5, "info": 6, "debug": 7} - - -def _tail_lines(path, n=400): - try: - with open(path, "rb") as f: - f.seek(0, 2) - size = f.tell() - f.seek(max(0, size - 256 * 1024)) - data = f.read() - except OSError: - return [] - return data.decode("utf-8", "replace").splitlines()[-n:] - - -def _wait_line(path, needle, timeout=6.0): - """Poll a log file for a line containing needle; the matched line or None.""" - deadline = time.time() + timeout - while time.time() < deadline: - for line in _tail_lines(path, 200): - if needle in line: - return line - time.sleep(0.25) - return None - - -@test("logs.routing", title="Log routing, relays, and the rendered rules", subsystem="logs", - kind="auto", est_min=1, - covers=_ROUTING_COVERS, - description="rsyslog is the only logger (no busybox syslogd/klogd), the rules rendered from " - "the settings and the effective-levels record exist and agree with /logs, every " - "logger has its directory, the daemon's own emitter and the per-controller " - "`logger` relay both land in the right file in the ff_line format, a stray " - "program's line goes to system/ and nowhere else, kernel lines reach kernel/, " - "and nothing of ours is written outside the tree.") -def routing(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - - # 1. one logger daemon - rs = hw.pidof("rsyslogd") - ev["rsyslogd_pids"] = rs - ctx.check(rs, "rsyslogd is not running") - for legacy in ("syslogd", "klogd"): - pids = hw.pidof(legacy) - ctx.check(not pids, "busybox %s is running (%s) next to rsyslog", legacy, pids) - ctx.log("rsyslogd %s; no syslogd/klogd", rs) - - # 2. the render step: rules + effective levels, consistent with what - # the API reports as effective - ctx.check(os.path.isfile("/data/forgefirm/rsyslog-forgefirm.conf"), - "rendered rules missing: /data/forgefirm/rsyslog-forgefirm.conf") - rules = _read("/data/forgefirm/rsyslog-forgefirm.conf") or "" - for name in LOGGERS: - ctx.check(("# %s:" % name) in rules, "rendered rules carry no block for %s", name) - ctx.check(os.path.isfile("/var/run/forgefirm-loglevels"), "effective-levels record missing") - st, body = fc.get("/logs") - ctx.check(st == 200 and isinstance(body, dict), "GET /logs -> %s", st) - ctx.check(body.get("effective_known") is True, "/logs does not know the effective levels") - eff = {} - for l in body.get("loggers") or []: - eff[l["name"]] = (l.get("effective_disk"), l.get("effective_remote")) - ev["effective"] = eff - ctx.log("effective levels: %s", eff) - ctx.check(set(eff) == set(LOGGERS), "loggers reported: %s", sorted(eff)) - levels = _read("/var/run/forgefirm-loglevels") or "" - for name, (disk, _remote) in eff.items(): - ctx.check(("log_%s_disk=%s" % (name, disk)) in levels, - "effective record and /logs disagree on %s (%s)", name, disk) - - # 3. the tree - for name in LOGGERS: - ctx.check(os.path.isdir(os.path.join(LOGS_ROOT, name)), "no directory for %s", name) - - # 4. the daemon's own emitter (fflog): a line from the running pid, if - # its level admits info - pids = hw.pidof("forgectrl") - ctx.check(pids, "forgectrl is not running") - disk_rank = _SEV_RANK.get(eff.get("forgectrl", ("info", "off"))[0], 6) - fpath = os.path.join(LOGS_ROOT, "forgectrl", "forgectrl.log") - if disk_rank >= 6: - own = [l for l in _tail_lines(fpath, 2000) - if any((" forgectrl[%d] " % p) in l for p in pids)] - ev["own_lines"] = len(own) - ctx.log("forgectrl[%s] lines in its file: %d", pids, len(own)) - ctx.check(own, "no line from the running daemon in %s (emitter path broken?)", fpath) - m = _LINE_RE.match(own[-1]) - ctx.check(m and m.group("prog") == "forgectrl", "not in the ff_line format: %r", own[-1]) - else: - ctx.log("forgectrl disk level %s: own-emitter check skipped", eff["forgectrl"][0]) - - # 5. the relay path: a `logger` probe under a controller's name lands - # in that controller's file with the ff_line format - nonce = "forgetest-routing-%d" % int(time.time()) - for name in ("grblhal", "gfhome"): - rank = _SEV_RANK.get(eff.get(name, ("info", "off"))[0], 6) - if rank < 3: - ctx.log("%s disk level off: relay probe skipped", name) - continue - rc, out = hw.run(["logger", "-t", name, "-p", "daemon.err", "%s %s" % (nonce, name)]) - ctx.check(rc == 0, "logger -t %s failed: %s", name, out) - path = os.path.join(LOGS_ROOT, name, name + ".log") - line = _wait_line(path, nonce) - ev["probe_" + name] = line - ctx.check(line, "%s probe did not reach %s", name, path) - m = _LINE_RE.match(line) - ctx.check(m and m.group("prog") == name and m.group("sev") == "ERR", - "%s probe line not in the ff_line format: %r", name, line) - ctx.log("%s: %s", name, line) - # a probe under another name must NOT leak into these files - rc, out = hw.run(["logger", "-t", "forgetest-stray", "-p", "daemon.err", "%s stray" % nonce]) - time.sleep(1.5) - for name in ("grblhal", "gfhome", "forgectrl", "forgeext"): - path = os.path.join(LOGS_ROOT, name, name + ".log") - ctx.check(not any((nonce + " stray") in l for l in _tail_lines(path, 100)), - "a stray program's line landed in %s", path) - sys_rank = _SEV_RANK.get(eff.get("system", ("info", "off"))[0], 6) - if sys_rank >= 3: - line = _wait_line(os.path.join(LOGS_ROOT, "system", "system.log"), nonce + " stray") - ctx.check(line, "the stray program's line did not reach system/") - - # 6. relay processes: the supervisor spawns one `logger` per controller - # (its stdin is the controller's output pipe); the daemon's init - # script keeps one on the fifo - st, mode = fc.get("/mode") - relays = hw.pidof("logger") - ev["logger_relays"] = relays - ev["controller"] = mode.get("controller") if isinstance(mode, dict) else None - ctx.log("logger relays: %s (controller %s)", relays, ev["controller"]) - if isinstance(mode, dict) and mode.get("controller") == "running": - ctx.check(relays, "controller running but no `logger` relay process") - if os.path.exists("/var/run/forgectrl.stderr"): - ctx.check(len(relays) >= 2 or ev["controller"] != "running", - "fifo relay present but only %s logger process(es)", relays) - - # 7. the kernel reaches kernel/ (unless filtered off) - if _SEV_RANK.get(eff.get("kernel", ("info", "off"))[0], 6) >= 6: - klines = _tail_lines(os.path.join(LOGS_ROOT, "kernel", "kernel.log"), 50) - ev["kernel_lines"] = len(klines) - ctx.check(klines and any(" kernel[" in l for l in klines), "no kernel lines in kernel/") - - # 8. nothing outside the tree - for stray in ("/data/forgectrl.log", "/data/gfcloud.log", "/data/gfhome.log", - "/data/log/gfcloud", "/data/log/gfhome", "/var/log/messages"): - ctx.check(not os.path.exists(stray), "pre-syslog log path still present: %s", stray) - ctx.log("no ForgeFIRM log outside %s", LOGS_ROOT) diff --git a/forgetest/forgetest/suite/setup.py b/forgetest/forgetest/suite/setup.py index ccfff41..e7b6264 100644 --- a/forgetest/forgetest/suite/setup.py +++ b/forgetest/forgetest/suite/setup.py @@ -537,121 +537,6 @@ def login_checks(ctx, name, pw): ctx.log("session set by the login, seen by GET /wiz, ended by the logout") -# ------------------------------------------------------------------ ssh - -def sshd_policy(): - """The ForgeFIRM keys of the effective sshd policy (sshd -T, - lowercase keys), or {"error": ...} when sshd cannot report it.""" - for exe in ("/usr/sbin/sshd", "sshd"): - rc, out = hw.run([exe, "-T"], timeout=15) - if rc == 127 and "No such file" in out: - continue - if rc != 0: - return {"error": out.strip()[:200]} - policy = {} - for line in out.splitlines(): - parts = line.split(None, 1) - if len(parts) == 2 and parts[0] in ("permitrootlogin", "permitemptypasswords", - "passwordauthentication", "banner"): - policy[parts[0]] = parts[1].strip() - return policy - return {"error": "no sshd binary"} - - -def restore_ssh(fc, before): - """Put the SSH state back as found: the flag file, then sshd.""" - flag = ssh_flag_path() - if before.get("running") and not tcp_open("127.0.0.1", 22): - fc.post("/system/ssh", params={"enable": "1"}) - elif not before.get("running") and tcp_open("127.0.0.1", 22): - fc.post("/system/ssh", params={"enable": "0"}) - write_file(flag, b"" if before.get("enabled") else None) - - -@test("setup.ssh-until-reboot", title="SSH is switched from the panel until the next reboot", - subsystem="setup", kind="auto", est_min=1, - covers=[("forgectrl", "src/main.c"), ("forgectrl", "src/paths.h")], - requires=["forgectrl.auth"], - description="GET /system/ssh reports enabled, running, and dev_image, and agrees with the " - "flag file and port 22. POST enable=1 writes /run/forgefirm/ssh-enabled and " - "sshd listens on 22, with the effective policy (sshd -T) at PasswordAuthentication " - "yes, and on a release image PermitRootLogin no and PermitEmptyPasswords no (the " - "dev image turns root over SSH back on for the bench; the release gate checks " - "the release image's sshd_config as built), and with no pre-authentication " - "banner, so the machine says nothing to a client that has not logged in. " - "An enable outside 0 and 1 is refused (400). On a release " - "image POST enable=0 removes the flag and stops sshd; on the dev image the boot " - "rule keeps sshd up and the stop is never gated, so the test removes the flag " - "by hand (as a reboot does, the flag is on tmpfs) and sshd stays. The prior " - "state is restored.") -def ssh_until_reboot(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - st, before = fc.get("/system/ssh") - ctx.log("GET /system/ssh -> %s %s", st, before) - ctx.check(st == 200 and isinstance(before, dict), "GET /system/ssh -> %s", st) - for k in ("enabled", "running", "dev_image"): - ctx.check(isinstance(before.get(k), bool), "/system/ssh lacks a boolean %s", k) - ev["before"] = before - flag = ssh_flag_path() - ctx.check(os.path.exists(flag) == before["enabled"], "the flag file %s disagrees with enabled=%s", - flag, before["enabled"]) - ctx.check(tcp_open("127.0.0.1", 22) == before["running"], "port 22 disagrees with running=%s", - before["running"]) - try: - st, on = fc.post("/system/ssh", params={"enable": "1"}) - ev["enable"] = on - ctx.log("POST /system/ssh enable=1 -> %s %s", st, on) - ctx.check(st == 200 and isinstance(on, dict) and on.get("enabled") is True, - "enable=1 -> %s %s", st, on) - ctx.check(os.path.exists(flag), "enable=1 did not write %s", flag) - took = ctx.wait_for(lambda: tcp_open("127.0.0.1", 22), 15) - ctx.check(took is not None, "sshd does not listen on 22 within 15 s of enable=1") - ctx.log("sshd listens on 22 (%.1f s)", took) - policy = sshd_policy() - ev["sshd_policy"] = policy - ctx.log("sshd -T: %s", policy) - ctx.check(policy.get("passwordauthentication") == "yes", - "PasswordAuthentication is %r: the account cannot log in", policy.get("passwordauthentication")) - - # No pre-authentication banner: the machine says nothing to a - # client that has not logged in. The mark and the version go in - # the motd, which a login prints (image.health). - ctx.check(policy.get("banner") in (None, "none"), - "sshd sends a pre-authentication banner (%r)", policy.get("banner")) - - if not before["dev_image"]: - ctx.check(policy.get("permitrootlogin") == "no" and policy.get("permitemptypasswords") == "no", - "a release image runs sshd with PermitRootLogin %r, PermitEmptyPasswords %r", - policy.get("permitrootlogin"), policy.get("permitemptypasswords")) - - st, body = fc.post("/system/ssh", params={"enable": "2"}) - ctx.check(st == 400, "enable=2 -> %s, expected 400", st) - - if before["dev_image"]: - os.remove(flag) - st, off = fc.get("/system/ssh") - ctx.log("dev image: flag removed by hand; GET /system/ssh -> %s %s", st, off) - ctx.check(isinstance(off, dict) and off.get("enabled") is False and off.get("running") is True, - "with the flag gone the dev image reads %s", off) - else: - st, off = fc.post("/system/ssh", params={"enable": "0"}) - ctx.log("POST /system/ssh enable=0 -> %s %s", st, off) - ctx.check(st == 200 and isinstance(off, dict) and off.get("enabled") is False, - "enable=0 -> %s %s", st, off) - ctx.check(not os.path.exists(flag), "enable=0 left %s", flag) - took = ctx.wait_for(lambda: not tcp_open("127.0.0.1", 22), 15) - ctx.check(took is not None, "sshd still listens on 22 within 15 s of enable=0") - ev["after"] = off - finally: - restore_ssh(fc, before) - st, final = fc.get("/system/ssh") - ev["final"] = final - ctx.check(isinstance(final, dict) and final.get("enabled") == before["enabled"] - and final.get("running") == before["running"], - "the SSH state is not as found: %s (was %s)", final, before) - - # --------------------------------------------------- the machine's name def mac_suffix(): @@ -735,267 +620,3 @@ def machine_name(ctx): ev["avahi_pids"] = hw.pidof("avahi-daemon") ctx.check(not ev["avahi_pids"], "an mDNS responder is running: %s", ev["avahi_pids"]) ctx.check(not os.path.exists("/etc/avahi"), "/etc/avahi is on the image") - - -# --------------------------------------------------------- the first run - -def led_pattern(): - """The pattern the button LEDs hold, from the ledtrig_smooth - attributes: {"target": [r, g, b], "pulse_on": [...], "pulse_off": [...]} - (None where unreadable). The hardware breathes on its own once - pulse_on is set, so the pattern is the evidence, not a sample.""" - out = {"target": [], "pulse_on": [], "pulse_off": []} - for name in hw.BUTTON_LEDS: - for attr in out: - try: - with open(hw.leds_root() + name + "/" + attr) as f: - out[attr].append(int(f.read().strip())) - except (OSError, ValueError): - out[attr].append(None) - return out - - -def breathes_teal(pattern): - """True for the acceptance cue: red off, green and blue on, and the - green and blue channels pulsing.""" - t = pattern.get("target") or [] - on = pattern.get("pulse_on") or [] - if len(t) < 3 or len(on) < 3: - return False - return (t[0] or 0) == 0 and (t[1] or 0) > 0 and (t[2] or 0) > 0 and (on[1] or 0) > 0 and (on[2] or 0) > 0 - - -FIRST_RUN_SETTINGS = ["cloud_enabled", "homing_mode", "controller_mode", "gfcloud_home_timeout_s", - "cool_tec_present", "ui_units", "wifi_country"] - - -def first_run_record(status, base): - """A record that reads as a first run with the checks and the burns - behind it: no consent, no press, no account, no form wizard, not - complete; every check and burn of the catalog kept from `base` (the - real record) or written at its catalog version. It is the machine - after the button-hold reset of the consent and the account, so the - run walks the form steps and not the hour of checks and burns a - fresh machine is offered.""" - base = base or {} - ts = now_iso() - rec = {"schema": 1, "created": base.get("created") or ts, "advisories": {}, "wizards": {}, - "flags": dict(base.get("flags") or {})} - have = base.get("wizards") or {} - for w in status.get("wizards") or []: - if w.get("class") == "form": - continue - rec["wizards"][w["id"]] = have.get(w["id"]) or { - "version": int(w.get("version") or 1), "completed": ts, "result": {}, "applied": {}} - return rec - - -@contextlib.contextmanager -def first_run(ctx): - """The machine as a first run for the block: the record seeded by - first_run_record and the account record removed, under a forgectrl - restart; the settings the form steps write put back as found; the - real records restored under another restart, the system accounts - replayed from them, and the temporary home directory removed. Yields - a dict the block fills with `temp_account`.""" - fc = ctx.forgectrl - ev = ctx.evidence - before = wiz(fc) - ev["before"] = wiz_summary(before) - ev["account_before"] = (before.get("users") or {}).get("name") - try: - base = json.loads((read_file(record_path()) or b"{}").decode("utf-8")) - except ValueError: - base = {} - seed = json.dumps(first_run_record(before, base), indent=1).encode("utf-8") - homes = "/data/forgefirm/home" - try: - homes_before = set(os.listdir(homes)) - except OSError: - homes_before = set() - state = {"temp_account": None} - try: - with Restore(ctx, FIRST_RUN_SETTINGS): - with installed(ctx, {record_path(): seed, users_path(): None}): - w = wiz(fc) - ev["fresh"] = wiz_summary(w) - ctx.check(w.get("first_run") is True and not (w.get("users") or {}).get("exists") - and w.get("advisories_complete") is False and w.get("acceptance_done") is False, - "the machine does not read as a first run: %s", wiz_summary(w)) - m = mode(fc) - ctx.check(m.get("controller") == "gated", "the controller is %s on a first run", - m.get("controller")) - yield state - finally: - # the real records are back, whether or not the block passed; the - # system accounts follow the record, and a temporary account the - # block created goes with its home - rc, out = hw.initd("forgefirm-users", "reload") - ev["users_reload_rc"] = rc - ctx.log("forgefirm-users reload -> rc %s %s", rc, out.strip()[:200]) - temp = state.get("temp_account") - if temp and temp not in homes_before and temp != ev["account_before"]: - import shutil - shutil.rmtree(os.path.join(homes, temp), ignore_errors=True) - ctx.log("removed the temporary home directory of %r", temp) - after = wiz(fc) - ev["after"] = wiz_summary(after) - ctx.check((after.get("users") or {}).get("name") == ev["account_before"], - "the account reads %r after the restore, was %r", (after.get("users") or {}).get("name"), - ev["account_before"]) - ctx.check(after.get("completed") == before.get("completed") and after.get("gate") == before.get("gate"), - "the machine does not read as before: %s", wiz_summary(after)) - - -def finished_first_run(ctx): - """The end of a first run: the record complete, the gate open, the - controller running.""" - fc = ctx.forgectrl - ev = ctx.evidence - w = wiz(fc) - ev["completed"] = wiz_summary(w) - ctx.check(w.get("completed") is True and w.get("gate") == "open", - "the run did not complete with the gate open: %s", wiz_summary(w)) - m = wait_settled(ctx) - ev["mode_after_setup"] = m - ctx.check(isinstance(m, dict) and m.get("controller") == "running", - "the controller did not come back after the setup: %s", m) - - -@test("setup.first-run-flow", title="The first run, driven as the page drives it", subsystem="setup", - kind="operator", hardware="takeover", est_min=3, - covers=[("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), ("forgectrl", "src/users.*"), - ("forgectrl", "src/session.*"), ("forgectrl", "src/button.*"), ("forgectrl", "src/led.*"), - ("forgectrl", "src/sheetid.*"), ("forgectrl", "src/advisories.*"), ("forgectrl", "src/hooks.h"), - ("forgectrl", "src/main.c"), ("forgectrl", "docs/advisories/safety-and-risk.md"), - ("forgectrl", "docs/advisories/licenses.md"), ("forgectrl", "docs/advisories/privacy.md"), - ("forgectrl", "docs/advisories/cloud-service.md")], - actions=["button"], - description="With the record seeded as a first run behind the checks and no account, the " - "test makes the page's own calls in the page's order: POST /wiz/advisories/accept " - "for each document at the hash GET /wiz lists (the phrase for the typed one), " - "advisories_complete; POST /wiz/advisories/press, the button state waiting and " - "the button LEDs breathing teal, the press (the bench actuator's), " - "acceptance_done; POST /wiz/account with a temporary name and password, the " - "account exists and the response sets the session cookie, and the gate holds " - "with the reason 'setup required: preferences, machine, cloud' and no " - "controller (the seed keeps the coolant flow calibration done). Then the " - "required form steps, the gate's reason naming what is left " - "after each: POST /wiz/preferences with the units as found; POST /wiz/machine " - "with the model as found (basic when unknown) and no TEC; POST /wiz/cloud as " - "found (enabled with the phrase and the homing as found, or off). With the " - "intake fans counted (the machine's setting, which the seed leaves alone) the " - "gate opens and the supervisor starts the controller. POST /wiz/complete, which " - "Start using the machine and Finish later both call: the record complete, the " - "gate open, the controller running. The settings " - "the steps write, the real record and the account are restored, the system " - "accounts replayed, the temporary home removed. Nothing moves and the laser is " - "not involved.") -def first_run_flow(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - settings = fc.settings() or {} - with first_run(ctx) as state: - w = wiz(fc) - for d in w.get("documents") or []: - data = {"doc": d.get("id"), "hash": d.get("hash")} - if d.get("consent") == "typed": - data["phrase"] = d.get("phrase") or SAFETY_PHRASE - st, body = fc.post("/wiz/advisories/accept", data=data) - ctx.check(st == 200, "accept %s -> %s %s", d.get("id"), st, body) - w = wiz(fc) - ctx.check(w.get("advisories_complete") is True, "the documents do not read accepted: %s", wiz_summary(w)) - ctx.log("the four documents are accepted") - - st, body = fc.post("/wiz/advisories/press") - ctx.check(st == 200, "POST /wiz/advisories/press -> %s %s", st, body) - took = ctx.wait_for(lambda: wiz(fc).get("button") == "waiting", 30) - ctx.check(took is not None, "the press step did not start") - took = ctx.wait_for(lambda: breathes_teal(led_pattern()), 10) - pattern = led_pattern() - ev["led_pattern"] = pattern - ctx.check(took is not None, "the button LEDs do not breathe teal at the press step: %s", pattern) - # The daemon records the acceptance when the press status is polled - # with the button read as pressed (button_take_pressed), which the - # page does from its press step; the test polls the same way. - ctx.act("button", "press", until=lambda: press_status(fc).get("accepted") is True, timeout=600, - text="It breathes teal: this press is the acceptance.") - ctx.check(wiz(fc).get("acceptance_done") is True, "the press was not accepted") - ctx.log("the press step: teal (%s), the press accepted", pattern) - - name = "bench-%04x" % random.randrange(1 << 16) - pw = "flow-%08x" % random.randrange(1 << 32) - st, body, hdrs = request(fc.base, "POST", "/wiz/account", - data={"name": name, "password": pw}, - headers={"X-ForgeFIRM-Token": fc.token}) - ev["account"] = {"status": st, "cookie": bool(hdrs.get("set-cookie"))} - ctx.check(st == 200, "POST /wiz/account -> %s %s", st, body[:200]) - state["temp_account"] = name - ctx.check(session_from_cookie(hdrs.get("set-cookie")) is not None, - "the account step did not log this client in: %r", hdrs.get("set-cookie")) - w = wiz(fc) - ctx.check((w.get("users") or {}).get("exists") is True and (w.get("users") or {}).get("name") == name, - "the account does not read created: %s", w.get("users")) - ctx.log("account %r created, the session cookie set", name) - - # After the account the gate waits for the required steps: no - # controller yet, and the reason names what is left. - def still_gated(left): - w = wiz(fc) - m = mode(fc) - why = "setup required: " + ", ".join(left) - ev["gated " + why] = {"wiz": wiz_summary(w), "mode": m} - ctx.check(w.get("gate") == "closed" and w.get("why") == why, - "GET /wiz does not hold the gate on %s: %s", ", ".join(left), wiz_summary(w)) - ctx.check(m.get("controller") == "gated", "the controller is %s with %s left", m.get("controller"), - ", ".join(left)) - - still_gated(REQUIRED_FORM_STEPS) - ctx.log("the gate holds after the account: %s", wiz(fc).get("why")) - - st, body = fc.post("/wiz/preferences", data={"ui_units": settings.get("ui_units") or "metric"}) - ctx.check(st == 200, "POST /wiz/preferences -> %s %s", st, body if isinstance(body, dict) else "") - still_gated(REQUIRED_FORM_STEPS[1:]) - machine = (w.get("machine") or {}) - model = machine.get("model") if machine.get("model") in ("basic", "plus", "pro") else "basic" - st, body = fc.post("/wiz/machine", data={"model": model, "tec": "0"}) - ctx.check(st == 200, "POST /wiz/machine -> %s %s", st, body if isinstance(body, dict) else "") - # The machine facts the step records see the head: its hardware id - # is the module's head/info "id" line. - head = (wiz(fc).get("machine") or {}).get("head") or {} - ev["head"] = head - ctx.check(head.get("present") is True and bool(head.get("hw_id")), - "the machine facts do not see the head: %s", head) - still_gated(REQUIRED_FORM_STEPS[2:]) - if settings.get("cloud_enabled") == "1": - data = {"enabled": "1", "phrase": SAFETY_PHRASE, - "homing_mode": settings.get("homing_mode") or "gfcloud"} - else: - data = {"enabled": "0"} - st, body = fc.post("/wiz/cloud", data=data) - ev["cloud_step"] = {"data": data, "status": st} - ctx.check(st == 200, "POST /wiz/cloud %s -> %s %s", data, st, body if isinstance(body, dict) else "") - w = wiz(fc) - done = w.get("versions") or {} - for wid in FORM_WIZARDS: - ctx.check(done.get(wid) == 1, "the record does not carry %s at version 1: %s", wid, done) - ctx.log("preferences, machine (%s), cloud (%s) written", model, data.get("enabled")) - - # The required steps done and the intake fans counted (the - # machine's setting, which the seed leaves alone, so the page shows - # the Intake fans step as done): the gate opens and the controller - # runs, whatever the setup does next. - ev["after_required"] = wiz_summary(w) - ctx.check(w.get("gate") == "open" and not w.get("why") and w.get("fans_counted") is True, - "the gate does not open with the required steps: %s", wiz_summary(w)) - took = ctx.wait_for(lambda: mode(fc).get("controller") == "running", 60) - m = wait_settled(ctx) - ev["mode_after_required"] = m - ctx.check(took is not None and isinstance(m, dict) and m.get("controller") == "running", - "the controller did not start with the required steps: %s", m) - ctx.log("the gate is open; the controller runs (%.1f s)", took) - - st, body = fc.post("/wiz/complete") - ctx.check(st == 200, "POST /wiz/complete -> %s %s", st, body if isinstance(body, dict) else "") - finished_first_run(ctx) - ctx.log("the run completed: gate open, the controller running") diff --git a/forgetest/tests/test_manifest.py b/forgetest/tests/test_manifest.py index 6e5bac7..f294171 100644 --- a/forgetest/tests/test_manifest.py +++ b/forgetest/tests/test_manifest.py @@ -123,13 +123,14 @@ class CoverageReportTests(unittest.TestCase): def test_behavioral_paths_are_never_allowed_away(self): # An advisory document is behavior (the consent is to its hash): the allowlist's docs/** and - # **/*.md do not take it out of the report, as they do not take it out of a fingerprint. + # **/*.md do not take it out of the report, as they do not take it out of a fingerprint. No + # host-tested rows here: the real table's rows name the first-run documents. man = helpers.with_file(helpers.make_manifest(), "forgectrl", "docs/advisories/privacy.md", "text") rest = helpers.make_test("a.one", [("forgectrl", "src/**"), ("grblhal-glowforge", "**"), ("kernel-module-glowforge", "**"), ("linux-fslc", "**")]) - self.assertEqual(m.coverage_report(man, [rest]), {"forgectrl": ["docs/advisories/privacy.md"]}) + self.assertEqual(m.coverage_report(man, [rest], host=()), {"forgectrl": ["docs/advisories/privacy.md"]}) doc = helpers.make_test("a.two", [("forgectrl", "docs/advisories/privacy.md")]) - self.assertEqual(m.coverage_report(man, [rest, doc]), {}) + self.assertEqual(m.coverage_report(man, [rest, doc], host=()), {}) self.assertNotEqual(m.fingerprint(man, doc.covers), m.fingerprint(helpers.with_file(man, "forgectrl", "docs/advisories/privacy.md", "new"), doc.covers)) diff --git a/forgetest/tests/test_setup_suite.py b/forgetest/tests/test_setup_suite.py index e3516bc..15d4c60 100644 --- a/forgetest/tests/test_setup_suite.py +++ b/forgetest/tests/test_setup_suite.py @@ -5,8 +5,8 @@ """The setup.* suite on the host: the registration (ids, kinds, the takeover tests, the operator tests' hands), the record builders, the -machine's name, the cookie parsing, the LED cue, and the settle rule -for a gated supervisor.""" +machine's name, the cookie parsing, and the settle rule for a gated +supervisor.""" import json import os import shutil @@ -19,9 +19,7 @@ import helpers # noqa: E402 from forgetest import baseline, catalog # noqa: E402 from forgetest.suite import setup # noqa: E402 -IDS = ("setup.account-login", - "setup.ssh-until-reboot", "setup.machine-name", "setup.first-run-flow") -OPERATOR = ("setup.first-run-flow",) +IDS = ("setup.account-login", "setup.machine-name") class RegistrationTests(unittest.TestCase): @@ -31,14 +29,8 @@ class RegistrationTests(unittest.TestCase): def test_every_test_is_registered_with_its_kind(self): kinds = {tid: self.reg[tid].kind for tid in IDS if tid in self.reg} self.assertEqual(sorted(kinds), sorted(IDS)) - for tid in OPERATOR: - self.assertEqual(kinds[tid], "operator", tid) for tid in IDS: - if tid not in OPERATOR: - self.assertEqual(kinds[tid], "auto", tid) - - def test_the_record_swapping_tests_are_takeovers(self): - self.assertEqual(self.reg["setup.first-run-flow"].hardware, "takeover") + self.assertEqual(kinds[tid], "auto", tid) def test_recommended_reads_the_status_document(self): w = {"recommended": [{"id": "laser.floor", "reason": "the tube was replaced"}, @@ -60,18 +52,6 @@ class RegistrationTests(unittest.TestCase): from forgetest.suite import setup_dark self.assertIs(reg["setup.check-airflow"].precheck, setup_dark.intakes_counted) - def test_the_first_run_flow_runs_unattended_with_the_fixture(self): - flow = self.reg["setup.first-run-flow"] - self.assertEqual(list(flow.actions), ["button"]) - self.assertFalse(flow.hands) - self.assertTrue(flow.fixture_runnable(("button", "lid", "interlock"))) - - def test_the_first_run_flow_covers_the_backend(self): - flow = set(self.reg["setup.first-run-flow"].covers) - for want in (("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), ("forgectrl", "src/users.*"), - ("forgectrl", "src/button.*"), ("forgectrl", "src/led.*"), ("forgectrl", "src/advisories.*")): - self.assertIn(want, flow, want) - def test_the_only_attended_tests_need_a_workstation_or_a_scrap(self): # With the bench actuator up, two setup tests keep a person: the # print from the Glowforge app, and the focus burn on a fresh @@ -240,16 +220,6 @@ class SmallHelpersTests(unittest.TestCase): self.assertIsNone(setup.session_from_cookie("ffsid=; Path=/; Max-Age=0")) self.assertIsNone(setup.session_from_cookie("")) - def test_the_teal_breathe_cue(self): - self.assertTrue(setup.breathes_teal({"target": [0, 180, 200], "pulse_on": [0, 1400, 1400], - "pulse_off": [0, 1400, 1400]})) - self.assertFalse(setup.breathes_teal({"target": [0, 255, 40], "pulse_on": [0, 0, 0], - "pulse_off": [0, 0, 0]})) # solid green - self.assertFalse(setup.breathes_teal({"target": [200, 200, 200], "pulse_on": [1800, 1800, 1800], - "pulse_off": [1800] * 3})) # white - self.assertFalse(setup.breathes_teal({"target": [None, None, None], "pulse_on": [None] * 3})) - self.assertFalse(setup.breathes_teal({})) - def test_decode(self): self.assertEqual(setup.decode(b'{"a": 1}'), {"a": 1}) self.assertEqual(setup.decode(b"cloud mode is not enabled"), "cloud mode is not enabled") @@ -276,65 +246,5 @@ class GatedSettleTests(unittest.TestCase): self.assertTrue(any("controller=gated" in ln for ln in lines)) -class FirstRunSeedTests(unittest.TestCase): - """The record the first-run tests install: a first run behind the - checks and the burns.""" - - STATUS = {"wizards": [{"id": "advisories", "version": 1, "class": "form"}, - {"id": "account", "version": 1, "class": "form"}, - {"id": "cloud", "version": 1, "class": "form"}, - {"id": "switches", "version": 1, "class": "dark"}, - {"id": "motion", "version": 2, "class": "dark"}, - {"id": "laser.focus", "version": 1, "class": "live"}]} - - def test_no_consent_no_account_not_complete_and_every_check_done(self): - base = {"created": "2026-09-01T00:00:00Z", "completed": "2026-09-02T00:00:00Z", - "advisories": {"safety-and-risk": {"hash": "x"}}, "acceptance": {"pressed_at": "t"}, - "account": {"name": "scott", "uid": 1000}, "flags": {"flow_thin": True}, - "wizards": {"advisories": {"version": 1}, "switches": {"version": 1, "result": {"lid": True}}}} - rec = setup.first_run_record(self.STATUS, base) - self.assertNotIn("completed", rec) - self.assertNotIn("acceptance", rec) - self.assertNotIn("account", rec) - self.assertEqual(rec["advisories"], {}) - self.assertEqual(rec["created"], "2026-09-01T00:00:00Z") - self.assertEqual(rec["flags"], {"flow_thin": True}) - self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"]) - self.assertEqual(rec["wizards"]["switches"]["result"], {"lid": True}) # kept from the base - self.assertEqual(rec["wizards"]["motion"]["version"], 2) # written at the catalog version - - def test_without_a_base_record(self): - rec = setup.first_run_record(self.STATUS, None) - self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"]) - self.assertEqual(rec["wizards"]["laser.focus"]["version"], 1) - - -class SshdPolicyTests(unittest.TestCase): - def test_the_three_keys_out_of_sshd_t(self): - from forgetest import hw - calls = [] - - def fake_run(cmd, timeout=60): - calls.append(cmd) - return 0, "port 22\npermitrootlogin no\npasswordauthentication yes\npermitemptypasswords no\nx11forwarding no\n" - real = hw.run - hw.run = fake_run - try: - self.assertEqual(setup.sshd_policy(), {"permitrootlogin": "no", "passwordauthentication": "yes", - "permitemptypasswords": "no"}) - finally: - hw.run = real - self.assertEqual(calls, [["/usr/sbin/sshd", "-T"]]) - - def test_a_failing_sshd_reports_the_error(self): - from forgetest import hw - real = hw.run - hw.run = lambda cmd, timeout=60: (1, "sshd: no hostkeys available") - try: - self.assertEqual(setup.sshd_policy(), {"error": "sshd: no hostkeys available"}) - finally: - hw.run = real - - if __name__ == "__main__": unittest.main()