vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-15 17:19:55 -07:00
Code Issues Packages Projects Releases Wiki Activity

Catalog: the LAN redirect, the release check, and the update job's lease leave

Browse Source 
Three catalog tests leave for the host proofs that carry them; the
catalog goes from 95 to 92.

- setup.https-only-writes, for forgectrl's tests/host_lan.sh (13273a0):
  from a LAN peer in a network namespace, reads served over HTTPS and
  HTTP, three writes over HTTP each a 302 to the same path on HTTPS that
  acts on nothing, the same write served from loopback, and the token
  writing over HTTPS with no account.
- update.release-check and update.job-locks, together, for forgectrl's
  tests/host_update.sh (acc3dcf), against a release server of its own:
  GET /update/release's keys, POST /update/check with a v<semver> tag, a
  size above 0, notes and new; the dismissal refused for a version that
  is not a tag, taken for the tag, cleared by an empty one; and a probe
  download that holds the machine (a settings write and a G21 job each
  409 in the job's name, update:download of kind system the holder),
  ends refused by the signature check leaving no file, and lets the same
  settings write through after.

update.release-check was the last test in suite/update.py; _VERSION_RX
and the re import, which only it used, go with it. update.job-locks was
the only test in suite/updlock.py: the module and its import in
suite/__init__.py go. setup.https-only-writes was the only test of
setup.py's "HTTPS boundary" section. tests/test_setup_suite.py drops its
id.

update.product-gate stays: fwup is on neither this host's PATH nor CI's,
so host_update.sh does not prove the five uploads, and no HOST_TESTED
row is added for src/fwproduct.*. No other HOST_TESTED row either.

test_setup_suite.py passes (31 tests). The coverage lint on the tree
manifest finds no uncovered path and no host-tested row problem across
the 92 tests; its one finding is setup.check-lens's empty src/lensfind.*
entry. pyflakes reports only the unused local plain in
update.slots-and-signature, which HEAD already has. The full forgetest
suite was not run.

Fingerprints: setup.py's and update.py's shared text move, so 33 tests
move: the 6 left in setup.py, the 25 in the modules that read it, and
update.slots-and-signature and update.product-gate. The pending pin bump
from c274f19 already owes 31 of them; setup.machine-name and
exthost.platform are the two it adds, as before. releases/ and
TEST-AUDIT.md are unchanged.
This commit is contained in:
ScottW514
2026-10-14 12:06:11 -04:00
parent 21046cd6ff
commit b447ae77f1
5 changed files with 1 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/__init__.py
-1
View File
@@ -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
forgetest/forgetest/suite/setup.py
-71
View File
@@ -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():
forgetest/forgetest/suite/update.py
-74
View File
@@ -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<semver> 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<semver> 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=<not a tag> -> %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)
forgetest/forgetest/suite/updlock.py
-121
View File
@@ -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)
forgetest/tests/test_setup_suite.py
+1 -1
View File
@@ -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")