diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index cb13fa3..9ca6612 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -24,7 +24,6 @@ from . import exthost # noqa: F401,E402 from . import extcall # noqa: F401,E402 from . import extcat # noqa: F401,E402 from . import extwizard # noqa: F401,E402 -from . import updlock # noqa: F401,E402 from . import homeoff # noqa: F401,E402 from . import bedsize # noqa: F401,E402 from . import extsender # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/setup.py b/forgetest/forgetest/suite/setup.py index 7fc7527..2e44916 100644 --- a/forgetest/forgetest/suite/setup.py +++ b/forgetest/forgetest/suite/setup.py @@ -537,77 +537,6 @@ def login_checks(ctx, name, pw): ctx.log("session set by the login, seen by GET /wiz, ended by the logout") -# ------------------------------------------------------- HTTPS boundary - -@test("setup.https-only-writes", title="Writes from the LAN go to HTTPS", subsystem="setup", - kind="auto", est_min=1, - covers=[("forgectrl", "src/auth.*"), ("forgectrl", "src/tls.*"), ("forgectrl", "src/peer.*"), - ("forgectrl", "src/main.c")], - requires=["forgectrl.auth"], - description="From the board's own LAN address, which the daemon sees as a non-loopback peer: " - "a read over HTTP is served (200), a write over HTTP is a 302 to the same path " - "on HTTPS and acts on nothing (the reboot route included), the same write from " - "loopback over HTTP is served, and over HTTPS from the LAN address the token " - "writes on the dev image (a release image needs the session: 403 login " - "required).") -def https_only_writes(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ip = lan_ip() - ev["lan_ip"] = ip - ctx.check(ip, "cannot determine the board's LAN address") - u = urllib.parse.urlsplit(fc.base) - http_base = "http://%s%s" % (ip, (":%d" % u.port) if u.port else "") - tls = "https://%s" % ip - token = {"X-ForgeFIRM-Token": fc.token} - - # the HTTPS listener first: without it the daemon serves every route - # on HTTP and the probes below would be served, not redirected - st, body, hdrs = request(tls, "GET", "/status") - ev["lan_https_read"] = st - ctx.log("GET %s/status -> %s", tls, st) - ctx.check(st == 200, "a read over HTTPS from the LAN -> %s: no HTTPS listener", st) - - st, body, hdrs = request(http_base, "GET", "/status") - ev["lan_http_read"] = st - ctx.log("GET %s/status -> %s", http_base, st) - ctx.check(st == 200, "a read over HTTP from the LAN -> %s, expected 200", st) - - # the probes act on nothing even when served: a settings write of the - # value in force, and two write-class reads - settings = fc.settings() - key = "ui_units" - val = settings.get(key) if isinstance(settings.get(key), str) and settings.get(key) else "metric" - for method, path, data in (("POST", "/settings", {key: val}), ("GET", "/system/ssh", None), - ("GET", "/logs", None)): - st, body, hdrs = request(http_base, method, path, data=data, headers=token) - loc = hdrs.get("location") or "" - ev["lan_http %s %s" % (method, path)] = {"status": st, "location": loc} - ctx.log("%s %s%s (token) -> %s Location %s", method, http_base, path, st, loc) - ctx.check(st == 302, "%s %s over HTTP from the LAN -> %s, expected 302", method, path, st) - ctx.check(loc.startswith("https://%s" % ip) and loc.endswith(path), - "the redirect does not point at %s on HTTPS: %r", path, loc) - - st, body = fc.post("/settings", data={key: val}) - ev["loopback_http_write"] = st - ctx.check(st == 200, "the same write from loopback over HTTP -> %s, expected 200", st) - - st, ssh = fc.get("/system/ssh") - dev = bool(isinstance(ssh, dict) and ssh.get("dev_image")) - account = bool((wiz(fc).get("users") or {}).get("exists")) - ev["dev_image"] = dev - ev["account"] = account - st, body, hdrs = request(tls, "POST", "/settings", data={key: val}, headers=token) - ev["lan_https_write"] = st - ctx.log("POST %s/settings (token, no session) -> %s (dev image %s, account %s)", tls, st, dev, account) - if dev or not account: - ctx.check(st == 200, "the token write over HTTPS -> %s, expected 200 on the dev image", st) - else: - b = decode(body) - ctx.check(st == 403 and isinstance(b, dict) and b.get("error") == "login required", - "the token write over HTTPS without a session -> %s %r, expected 403 login required", st, b) - - # ------------------------------------------------------------------ ssh def sshd_policy(): diff --git a/forgetest/forgetest/suite/update.py b/forgetest/forgetest/suite/update.py index 32ce59f..cace040 100644 --- a/forgetest/forgetest/suite/update.py +++ b/forgetest/forgetest/suite/update.py @@ -5,7 +5,6 @@ """update.* - the A/B slot inventory and the firmware verification path.""" import os -import re import tempfile import time import shutil @@ -230,76 +229,3 @@ def product_gate(ctx): shutil.rmtree(work, ignore_errors=True) ctx.log("PASS: three extension packages and one foreign product refused at the firmware upload with " "nothing staged; firmware taken as the unsigned upload it is") - - -_VERSION_RX = re.compile(r"^v?\d+\.\d+\.\d+") - - -@test("update.release-check", title="Release check answers from the releases API; the alert dismissal is per release", - subsystem="update", kind="auto", est_min=1, - covers=_UPDATE_COVERS + [("forgectrl", "src/ui/panel.js"), ("forgectrl", "src/ui/index.html")], - requires=["forgectrl.auth"], - description="GET /update/release answers the kept answer of the daily check (available, version, " - "current, new, checked, dismissed). POST /update/check asks the releases API now and " - "answers the same shape; a machine with no route to the API answers 502, which the drill " - "records and steps over. A published release is a v tag with the firmware " - "file's size and its notes, and `new` follows the version order: a development build " - "sees every release as newer. POST /update/dismiss marks that version dismissed, an " - "empty version undoes it, and a version that is not a tag is refused. The dismissal " - "the machine had before the drill is put back.") -def release_check(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - keys = ("available", "version", "current", "new", "checked", "dismissed", "detail", "bytes") - - st, before = fc.get("/update/release") - ctx.log("GET /update/release -> %s %s", st, {k: before.get(k) for k in keys} if isinstance(before, dict) else before) - ctx.check(st == 200 and isinstance(before, dict) and "available" in before and "checked" in before, - "GET /update/release -> %s", st) - ev["before"] = {k: before.get(k) for k in keys} - prior = before.get("version", "") if before.get("dismissed") else "" - - st, now = fc.post("/update/check") - ctx.log("POST /update/check -> %s %s", st, {k: now.get(k) for k in keys} if isinstance(now, dict) else now) - if st == 502: - ctx.log("the machine has no route to the releases API; the kept answer stands") - now = before - else: - ctx.check(st == 200 and isinstance(now, dict) and "available" in now, "POST /update/check -> %s", st) - ctx.check(isinstance(now.get("checked"), int) and now["checked"] >= before.get("checked", 0), - "the check did not move `checked`: %s", now.get("checked")) - ev["after_check"] = {k: now.get(k) for k in keys} - - if now.get("available"): - ctx.check(_VERSION_RX.match(str(now.get("version", ""))) is not None, - "the release version is not a v tag: %r", now.get("version")) - ctx.check(isinstance(now.get("bytes"), int) and now["bytes"] > 0, - "the firmware file's size is missing: %r", now.get("bytes")) - ctx.check(isinstance(now.get("notes"), str), "the notes are not a string") - ctx.check(isinstance(now.get("new"), bool), "`new` is not a bool") - if not _VERSION_RX.match(str(now.get("current", ""))): - ctx.check(now.get("new") is True, - "a development build (%r) must see release %s as newer", now.get("current"), now["version"]) - else: - ctx.log("no release available: %s", now.get("detail")) - ctx.check(isinstance(now.get("detail"), str) and now["detail"], "an unavailable release names no reason") - - try: - st, body = fc.post("/update/dismiss", params={"version": "v0.0.1; rm -rf /"}) - ctx.log("POST /update/dismiss version= -> %s %s", st, body) - ctx.check(st == 400, "a version that is not a tag was not refused: %s", st) - - tag = now["version"] if now.get("available") else "v0.0.0" - st, body = fc.post("/update/dismiss", params={"version": tag}) - ctx.log("POST /update/dismiss version=%s -> %s %s", tag, st, - {k: body.get(k) for k in keys} if isinstance(body, dict) else body) - ctx.check(st == 200 and isinstance(body, dict), "POST /update/dismiss -> %s", st) - if now.get("available"): - ctx.check(body.get("dismissed") is True, "the dismissal of %s was not recorded: %r", tag, body.get("dismissed")) - st, body = fc.post("/update/dismiss", params={"version": ""}) - ctx.log("POST /update/dismiss version= -> %s", st) - ctx.check(st == 200 and isinstance(body, dict) and body.get("dismissed") is False, - "the empty version did not undo the dismissal: %s %r", st, body.get("dismissed") if isinstance(body, dict) else body) - finally: - st, body = fc.post("/update/dismiss", params={"version": prior}) - ctx.log("dismissal put back to %r -> %s", prior, st) diff --git a/forgetest/forgetest/suite/updlock.py b/forgetest/forgetest/suite/updlock.py deleted file mode 100644 index 9ab0e44..0000000 --- a/forgetest/forgetest/suite/updlock.py +++ /dev/null @@ -1,121 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""An update job holds the machine, on the machine. - -Its own module, so that no other test's fingerprint moves. The job is a -probe download of the published release: the download job itself, run on -the release's acceptance record instead of its firmware file, because -GitHub counts every request of the firmware file as an install. The -verification refuses the record and the probe keeps nothing; a download -the machine had staged is left as it was. -""" - -import os - -from ..catalog import test -from .motion import _job_post, _words - -DL_FW = "/data/forgefirm/download.fw" -DL_PROBE = "/data/forgefirm/download-probe" -HELD = "an update job (download) holds the machine" - - -def _as_found(path): - try: - st = os.stat(path) - return [st.st_size, st.st_mtime_ns] - except OSError: - return None - - -@test("update.job-locks", title="An update job holds the machine: the settings are locked while it runs", - subsystem="update", kind="auto", est_min=2, - covers=[("forgectrl", "src/update.*"), ("forgectrl", "src/relcheck.*"), ("forgectrl", "src/lease.*")], - requires=["update.release-check"], - description="With a release published on the releases API, POST /update/download?probe=1 starts the " - "download job (202) on the release's acceptance record in place of its firmware file, so " - "the test adds nothing to the firmware file's download count. The machine lease is taken " - "before the 202: a settings write and a posted job right after it are each refused in the " - "job's name (409, an update job (download) holds the machine, the settings locked), and " - "/status names the job (update:download, of kind system) as the holder while it runs. The " - "job fetches the record through the download's own path and ends refused by the signature " - "check with the record discarded. When it ends, the lease is free and the same settings " - "write is taken. A download the machine had staged is untouched and the probe leaves no " - "file. That a log export does not lock the settings while every other holder does is " - "lease_test's; that a probe never names the firmware file is relcheck_test's.") -def job_locks(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle") - st, rel = fc.get("/update/release") - if st == 200 and isinstance(rel, dict) and not rel.get("checked"): - # the kept answer does not outlive a restart of the daemon: ask now - st, rel = fc.post("/update/check") - ctx.log("the release was not checked since the daemon started: POST /update/check -> %s", st) - ev["release"] = {k: (rel or {}).get(k) for k in ("available", "version")} if isinstance(rel, dict) else rel - ctx.check(st == 200 and isinstance(rel, dict) and rel.get("available"), "no release is published to probe: %s", rel) - staged = _as_found(DL_FW) - ev["staged_download"] = staged - ctx.check(_as_found(DL_PROBE) is None, "a probe's file is left over from before: %s", DL_PROBE) - units = fc.settings().get("ui_units") or "metric" - - def holder(): - return ((fc.status().get("lease") or {}).get("holder") or {}) - - def job(): - st_, j = fc.get("/update/status") - return j if isinstance(j, dict) else {} - - try: - # The record is small, so the job can end within a second. The - # lease is taken before the 202, so the settings write goes first - # and at once: it is the same value, so a probe that ended before - # it only costs another probe. The posted job goes only while the - # settings are refused in the job's name. - refused = None - for attempt in (1, 2, 3): - st, body = fc.post("/update/download", params={"probe": "1"}) - ctx.check(st == 202, "POST /update/download probe=1 -> %s %s", st, _words(body)[:200]) - st, body = fc.post("/settings", params={"ui_units": units}) - if st == 200: - ctx.wait_for(lambda: job().get("running") is False or None, 120, poll=0.5) - result = job().get("result") or {} - ctx.log("probe %d ended before the settings write was read (%s); again", attempt, result) - ctx.check("signature verification failed" in str(result.get("error")), - "the probe ended, and not refused by the signature check: %s", result) - continue - st2, body2 = _job_post(fc, "G21\n", name="beside-an-update") - held = holder() - refused = {"a settings write": [st, _words(body)[:200]], "a posted job": [st2, _words(body2)[:200]]} - break - ctx.check(refused is not None, "three probes each ended before a settings write could be refused") - ev["refused"] = refused - ev["holder"] = held - ctx.log("beside the probe: %s; the lease: %s", refused, held) - ctx.check(st == 409 and HELD in _words(body) and "settings are locked" in _words(body), - "a settings write beside the probe -> %s %s", st, _words(body)[:200]) - ctx.check(st2 == 409 and HELD in _words(body2), "a job beside the probe -> %s %s", st2, _words(body2)[:200]) - if held: - ctx.check(held.get("owner") == "update:download" and held.get("kind") == "system", - "the probe does not hold the machine as the download job: %s", held) - else: - ctx.log("the probe had ended before /status was read; its refusals name it") - ctx.wait_for(lambda: job().get("running") is False or None, 600, poll=1.0) - done = job() - ev["job"] = done - ctx.log("the probe ended: %s", done) - result = done.get("result") or {} - ctx.check(result.get("ok") is False and "signature verification failed" in str(result.get("error")), - "the probe did not end refused by the signature check (the record fetched and discarded): %s", - result) - ctx.check(not holder(), "the lease was not given back: %s", holder()) - st, body = fc.post("/settings", params={"ui_units": units}) - ctx.check(st == 200, "the same settings write after the job -> %s %s", st, _words(body)[:200]) - finally: - # The job runs on after a check that failed: its end is waited for. - ctx.wait_for(lambda: job().get("running") is False or None, 600, poll=2.0) - ctx.check(_as_found(DL_FW) == staged, "the staged download moved: %s, and %s before", _as_found(DL_FW), staged) - ctx.check(_as_found(DL_PROBE) is None, "the probe left its file: %s", DL_PROBE) diff --git a/forgetest/tests/test_setup_suite.py b/forgetest/tests/test_setup_suite.py index 02c92a1..d7ba5a6 100644 --- a/forgetest/tests/test_setup_suite.py +++ b/forgetest/tests/test_setup_suite.py @@ -17,7 +17,7 @@ import helpers from forgetest import baseline, catalog from forgetest.suite import setup -IDS = ("setup.account-login", "setup.https-only-writes", +IDS = ("setup.account-login", "setup.ssh-until-reboot", "setup.machine-name", "setup.first-run-flow", "setup.first-run-page", "setup.mirror")