diff --git a/forgetest/forgetest/manifest.py b/forgetest/forgetest/manifest.py index 87c9cf4..a47e65a 100644 --- a/forgetest/forgetest/manifest.py +++ b/forgetest/forgetest/manifest.py @@ -131,6 +131,12 @@ HOST_TESTED = [ # The M-code relay to the package that answers (the controller's side is # the null-sink mcode_test harness, under grblhal-glowforge's src/**). ("forgectrl", "src/mcode.*", "forgectrl:tests/mcode_test.c", "forgectrl:.github/workflows/build.yml"), + # The digest primitives (SHA-256, HMAC, the constant-time compare) that + # the advisories' hashes, the tokens and the camera key are made with. + ("forgectrl", "src/sha256.*", "forgectrl:tests/sha256_test.c", "forgectrl:.github/workflows/build.yml"), + # The built-in extensions' table: the cloud entry, its refusals while it + # is off, the sweep when it goes off, the list. + ("forgectrl", "src/builtin.*", "forgectrl:tests/builtin_test.c", "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/exthost.py b/forgetest/forgetest/suite/exthost.py index 426af2d..02b966b 100644 --- a/forgetest/forgetest/suite/exthost.py +++ b/forgetest/forgetest/suite/exthost.py @@ -888,8 +888,9 @@ def _as_found(ctx, fc, prior, raw, dir_mode, found_tree): @test("exthost.service", title="A package's service runs confined under the extension host", subsystem="exthost", kind="auto", hardware="takeover", est_min=5, - covers=[("forgeext", "**"), ("forgectrl", "src/main.c"), ("forgectrl", "src/logs.*")], - requires=["exthost.platform", "setup.extensions-consent"], + covers=[("forgeext", "**"), ("forgectrl", "src/main.c"), ("forgectrl", "src/logs.*"), + ("forgectrl", "docs/advisories/extensions.md")], + requires=["exthost.platform"], description="The extension host is one running process (/usr/bin/forgeext run) and the status " "file is its word, and with extensions off it runs nothing and says why. The test builds a reference " "package on the board, signs it with a key it makes there, and adds that key as an owner " diff --git a/forgetest/forgetest/suite/setup.py b/forgetest/forgetest/suite/setup.py index b52b78e..cac7b06 100644 --- a/forgetest/forgetest/suite/setup.py +++ b/forgetest/forgetest/suite/setup.py @@ -23,7 +23,6 @@ No coverage map names it: a change there makes every test necessary again. """ import contextlib -import hashlib import posixpath import http.client import json @@ -54,7 +53,6 @@ REQUIRED_FORM_STEPS = ("preferences", "machine", "cloud") REQUIRED_STEPS = REQUIRED_FORM_STEPS + ("cooling.flow",) # The setting that counts the intake fans: the gate holds without it. INTAKE_KEY = "cool_intake_fans" -ADVISORY_DOCS = ("safety-and-risk", "licenses", "privacy", "cloud-service") SAFETY_PHRASE = "I UNDERSTAND" LOGIN_FAILS = 5 LOGIN_LOCK_S = 30 @@ -351,240 +349,6 @@ def wait_settled(ctx): return ctx.takeover().wait_settled() -# ------------------------------------------------------------- prechecks - -def gate_is_open(): - """Precheck: the tests that close the gate start from an open one.""" - try: - w = wiz(hw.Forgectrl()) - except hw.HwError as e: - return "forgectrl unreachable: %s" % e - if w.get("gate") != "open": - return ("the controller gate is closed (%s); the test starts from an open gate" - % (w.get("why") or "no reason")) - return None - - -# -------------------------------------------------------------- the gate - -_GATE_COVERS = [("forgectrl", "src/setup.*"), ("forgectrl", "src/super.c"), - ("forgectrl", "src/main.c"), ("forgectrl", "src/paths.h")] - - -@test("setup.gate-blocks-controllers", - title="The setup gate holds the controller until the advisories, the account, the required " - "steps, and the intake fans", - subsystem="setup", kind="auto", hardware="takeover", mode="grbl", est_min=8, - covers=_GATE_COVERS, requires=["forgectrl.auth"], precheck=gate_is_open, - description="Four records in turn, each installed under a forgectrl restart. Without the " - "consent (no advisory accepted, no press), GET /mode reports the controller " - "gated with the reason naming the advisories, GET /wiz agrees, no controller " - "process runs and the Grbl port is closed; with the consent and without the " - "account, the same with the reason naming the account; with the consent and " - "the account and no step done, the same with the reason 'setup required: " - "preferences, machine, cloud, cooling.flow'. With those four done as well and " - "no flag, the " - "gate is open with no reason: the supervisor spawns the controller and the Grbl " - "port accepts. (Which steps the table requires, misses and recommends is " - "forgectrl's setup_test, on the image's own table.) On that record the intake " - "fan count is cleared: the gate closes live with the reason naming the intake " - "fans, GET /wiz reports them not counted, and the supervisor stops the " - "controller; the count written back as found opens the gate again and the " - "controller comes back. The real record is restored under another restart and " - "the machine reads as before.") -def gate_blocks_controllers(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - before = wiz(fc) - ev["before"] = wiz_summary(before) - ctx.log("before: gate %s, %d steps recommended", before.get("gate"), len(recommended(before))) - raw = read_file(record_path()) - ctx.check(raw, "no setup record at %s", record_path()) - try: - record = json.loads(raw.decode("utf-8")) - except ValueError as e: - ctx.fail("the setup record is not JSON: %s", e) - - def encoded(rec): - return json.dumps(rec, indent=1, sort_keys=True).encode() + b"\n" - - def gated(part, words): - """The gate closed on `part`: the supervisor and GET /wiz say so - in `words`, and no controller runs.""" - took = ctx.wait_for(lambda: mode(fc).get("controller") == "gated", 60) - m = mode(fc) - ev["gated_mode " + part] = m - ctx.log("/mode without the %s: %s (after %s s)", part, m, took) - ctx.check(m.get("controller") == "gated" and m.get("gated") is True, - "without the %s the supervisor did not report the gate: %s", part, m) - ctx.check(words in (m.get("why") or ""), "the reason does not name the %s: %r", part, m.get("why")) - w = wiz(fc) - ev["gated_wiz " + part] = wiz_summary(w) - ctx.check(w.get("gate") == "closed" and words in (w.get("why") or ""), - "GET /wiz does not agree with the closed gate: %s", wiz_summary(w)) - ctx.check(not hw.grbl_port_open(timeout=2), "the Grbl port answers while the gate is closed") - pids = hw.pidof("grblHAL_glowfor") + hw.pgrep_f("gfcloud.py") - ev["controller_pids " + part] = pids - ctx.check(not pids, "a controller process runs while the gate is closed: %s", pids) - ctx.log("gated on the %s: no controller process, the Grbl port is closed", part) - - def running(what): - """The gate open: the supervisor spawns the controller and its - port accepts.""" - # The supervisor spawns on its next pass; a gated machine counts - # as settled, so wait for the spawn before the settle wait. - took = ctx.wait_for(lambda: mode(fc).get("controller") == "running", 60) - m = wait_settled(ctx) - ev["open_mode " + what] = m - ctx.check(took is not None and isinstance(m, dict) and m.get("controller") == "running", - "the controller did not come back %s: %s", what, m) - # /mode reads running from the fork; the controller binds its port - # a few tens of milliseconds later - opened = ctx.wait_for(lambda: hw.grbl_port_open(timeout=1), 10, poll=0.05) - ev["port_open_after_s " + what] = opened - ctx.check(opened is not None, "the Grbl port is closed with the controller running") - ctx.log("%s: the controller is running, the Grbl port accepts after %.3f s", what, opened) - - with installed(ctx, {record_path(): encoded(record_without_consent(record))}): - gated("consent", "the advisories are not accepted") - with ctx.takeover(): - write_file(record_path(), encoded(record_without_account(record))) - gated("account", "no account exists") - with ctx.takeover(): - write_file(record_path(), encoded(record_without_wizards(record))) - gated("required steps", "setup required: " + ", ".join(REQUIRED_STEPS)) - with ctx.takeover(): - write_file(record_path(), encoded(record_with_required_steps(record))) - - # the consent, the account, the required steps, the intake fans - # counted: nothing else closes the gate - w = wiz(fc) - ev["open_wiz"] = wiz_summary(w) - ctx.check(w.get("gate") == "open" and not w.get("why") and w.get("fans_counted") is True, - "the consent, the account, and the required steps do not open the gate: %s", wiz_summary(w)) - running("with the required steps done") - - # the intake fan count holds the gate as the account does - with Restore(ctx, [INTAKE_KEY]) as count: - ctx.check(count.before.get(INTAKE_KEY), "%s is unset with the gate open", INTAKE_KEY) - st, body = fc.post("/settings", params={INTAKE_KEY: ""}) - ctx.check(st == 200, "clearing %s -> %s %s", INTAKE_KEY, st, body) - ctx.wait_for(lambda: wiz(fc).get("fans_counted") is False, 15) - ctx.check(wiz(fc).get("fans_counted") is False, - "GET /wiz reports the intake fans counted with %s unset", INTAKE_KEY) - gated("intake fan count", "the intake fans are not counted") - running("with the intake fans counted again") - ctx.check((fc.settings() or {}).get(INTAKE_KEY) == count.before.get(INTAKE_KEY), - "%s reads %r, was %r", INTAKE_KEY, (fc.settings() or {}).get(INTAKE_KEY), - count.before.get(INTAKE_KEY)) - - # the real record is back under a restart - after = wiz(fc) - ev["after"] = wiz_summary(after) - ctx.check(after.get("gate") == before.get("gate"), - "the gate reads %s after the restore, was %s", after.get("gate"), before.get("gate")) - ctx.check(recommended(after) == recommended(before), - "the recommended list changed: %s -> %s", recommended(before), recommended(after)) - m = mode(fc) - ev["after_mode"] = m - ctx.check(m.get("controller") == "running", "the controller is %s after the restore", m.get("controller")) - ctx.log("restored: gate %s, controller %s", after.get("gate"), m.get("controller")) - - -# ------------------------------------------------------------ advisories - -@test("setup.advisories-rehash", title="An advisory is accepted at its current hash only", - subsystem="setup", kind="auto", hardware="takeover", est_min=3, - covers=[("forgectrl", "src/advisories.*"), ("forgectrl", "src/setup.*"), - ("forgectrl", "src/wiz.*"), ("forgectrl", "src/sha256.*"), ("forgectrl", "src/main.c"), - ("forgectrl", "src/ui/md.js"), ("forgectrl", "src/ui/embed_docs.cmake"), - ("forgectrl", "docs/advisories/safety-and-risk.md"), ("forgectrl", "docs/advisories/licenses.md"), - ("forgectrl", "docs/advisories/privacy.md"), ("forgectrl", "docs/advisories/cloud-service.md")], - requires=["forgectrl.auth"], - description="GET /advisories/safety-and-risk serves the document as markdown with an ETag " - "equal to the SHA-256 of the body and to the hash GET /wiz lists; an unknown " - "document is refused (400); an accept with a stale hash is refused (409); the " - "typed document without its phrase is refused (400); the accept at the current " - "hash with the phrase is recorded (200), and any new acceptance clears the " - "press (acceptance_done false, the gate closed on the consent). The previous " - "record is put back under a forgectrl restart and the machine reads as before.") -def advisories_rehash(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - raw = read_file(record_path()) - ctx.check(raw, "no setup record at %s", record_path()) - before = wiz(fc) - ev["before"] = wiz_summary(before) - docs = {d.get("id"): d for d in before.get("documents") or []} - ctx.check(set(docs) >= set(ADVISORY_DOCS), "GET /wiz lists %s, expected %s", sorted(docs), ADVISORY_DOCS) - doc = docs["safety-and-risk"] - ctx.check(doc.get("consent") == "typed" and doc.get("phrase") == SAFETY_PHRASE, - "the safety document is not the typed one: %s", doc) - - st, body, hdrs = request(fc.base, "GET", "/advisories/safety-and-risk", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - ev["etag"] = etag - ev["content_type"] = hdrs.get("content-type") - ctx.log("GET /advisories/safety-and-risk -> %s, %d bytes, ETag %s", st, len(body), etag) - ctx.check(st == 200 and body, "GET /advisories/safety-and-risk -> %s", st) - ctx.check((hdrs.get("content-type") or "").startswith("text/markdown"), - "the document is not served as markdown: %r", hdrs.get("content-type")) - ctx.check(etag == hashlib.sha256(body).hexdigest(), "the ETag is not the SHA-256 of the body") - ctx.check(etag == doc.get("hash"), "the ETag %s differs from the hash GET /wiz lists %s", etag, doc.get("hash")) - st, body, hdrs = request(fc.base, "GET", "/advisories/no-such-document", headers={"Host": fc.host_header()}) - ctx.check(st == 404, "an unknown document -> %s, expected 404", st) - - st, body = fc.post("/wiz/advisories/accept", data={"doc": "no-such-document", "hash": etag}) - ev["unknown_doc"] = st - ctx.check(st == 400, "an accept of an unknown document -> %s, expected 400", st) - st, body = fc.post("/wiz/advisories/accept", - data={"doc": "safety-and-risk", "hash": "0" * 64, "phrase": SAFETY_PHRASE}) - ev["stale_hash"] = st - ctx.log("accept with a stale hash -> %s %s", st, body if isinstance(body, dict) else "") - ctx.check(st == 409, "an accept with a stale hash -> %s, expected 409", st) - st, body = fc.post("/wiz/advisories/accept", data={"doc": "safety-and-risk", "hash": etag}) - ev["no_phrase"] = st - ctx.check(st == 400, "the typed document without its phrase -> %s, expected 400", st) - w = wiz(fc) - ctx.check(w.get("acceptance_done") == before.get("acceptance_done"), - "a refused accept changed the press record") - - try: - st, body = fc.post("/wiz/advisories/accept", - data={"doc": "safety-and-risk", "hash": etag, "phrase": SAFETY_PHRASE}) - ev["accept"] = st - ctx.log("accept at the current hash with the phrase -> %s", st) - ctx.check(st == 200 and isinstance(body, dict), "the accept -> %s %r", st, body) - after = wiz(fc) - ev["after_accept"] = wiz_summary(after) - accepted = {d.get("id"): d.get("accepted") for d in after.get("documents") or []} - ctx.check(accepted.get("safety-and-risk") is True, "the document does not read accepted") - ctx.check(after.get("acceptance_done") is False, - "a new acceptance did not clear the press (acceptance_done %s)", after.get("acceptance_done")) - ctx.check(after.get("gate") == "closed" and "not accepted" in (after.get("why") or ""), - "the gate does not close on the cleared press: %s", wiz_summary(after)) - # the record on disk moved with it - try: - on_disk = json.loads((read_file(record_path()) or b"{}").decode("utf-8")) - except ValueError: - on_disk = {} - ctx.check("acceptance" not in on_disk and - (on_disk.get("advisories") or {}).get("safety-and-risk", {}).get("hash") == etag, - "the record on disk does not carry the acceptance") - ctx.log("acceptance_done cleared, gate %s: %s", after.get("gate"), after.get("why")) - finally: - with ctx.takeover(): - write_file(record_path(), raw) - ctx.log("the previous record is back under a restart") - - final = wiz(fc) - ev["after_restore"] = wiz_summary(final) - ctx.check(final.get("acceptance_done") == before.get("acceptance_done") - and final.get("gate") == before.get("gate") - and final.get("advisories_complete") == before.get("advisories_complete"), - "the machine does not read as before: %s", wiz_summary(final)) - - # -------------------------------------------------------------- the login @test("setup.account-login", title="The panel login over HTTPS", subsystem="setup", @@ -959,266 +723,6 @@ def ssh_until_reboot(ctx): "the SSH state is not as found: %s (was %s)", final, before) -# ---------------------------------------------------------------- cloud - -@test("setup.cloud-disabled-surface", title="Nothing points at the cloud while it is off", - subsystem="setup", kind="auto", mode="grbl", est_min=1, - covers=[("forgectrl", "src/main.c"), ("forgectrl", "src/super.c"), ("forgectrl", "src/settings.*"), - ("forgectrl", "src/wiz.c"), ("forgectrl", "src/hooks.h"), ("forgectrl", "src/builtin.*")], - requires=["forgectrl.settings-bounds"], - description="The test turns cloud mode off itself with one write and puts every setting " - "back as found. GET /extensions lists cloud mode as a built-in with its two " - "roles (homing gfcloud, controller cloud), enabled as cloud_enabled says and each " - "role active exactly when its setting selects it. cloud_enabled=0 takes the " - "gfcloud homing and the cloud boot mode " - "down with it, as the cloud step does, and the list then reads off with no role " - "active. With it at 0, POST /settings " - "controller_mode=cloud and homing_mode=gfcloud are refused (409, in the words the " - "table gives each role) and leave the " - "settings unchanged, cloud_enabled=1 without the typed phrase is refused (400) " - "and leaves it at 0, and POST /mode controller=cloud is refused (409) with a " - "message that names cloud mode, the machine staying in GRBL mode.") -def cloud_disabled_surface(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - found = fc.settings() - prior = found.get("cloud_enabled") or "" - ev["found"] = {k: found.get(k) for k in ("cloud_enabled", "homing_mode", "controller_mode")} - - def cloud_entry(): - st, doc = fc.get("/extensions") - ctx.check(st == 200 and isinstance(doc, dict), "GET /extensions -> %s", st) - ext = [e for e in doc.get("extensions", []) if e.get("id") == "cloud"] - ctx.check(len(ext) == 1, "GET /extensions lists cloud mode %d times", len(ext)) - return ext[0] - - # The list agrees with the settings as found. - e = cloud_entry() - ev["extension_found"] = {"enabled": e.get("enabled"), "roles": e.get("roles")} - ctx.check(e.get("builtin") is True and e.get("enable_key") == "cloud_enabled" and e.get("tab") == "gfcloud" - and e.get("setup_step") == "cloud", "the cloud entry: %s", e) - ctx.check(e.get("enabled") is (prior == "1"), "the list says enabled=%s with cloud_enabled=%r", - e.get("enabled"), prior) - roles = {r["role"]: r for r in e.get("roles", [])} - ctx.check(set(roles) == {"homing", "controller"} and roles["homing"]["provider"] == "gfcloud" - and roles["homing"]["fallback"] == "none" and roles["controller"]["provider"] == "cloud" - and roles["controller"]["fallback"] == "grbl", "the cloud entry's roles: %s", e.get("roles")) - for r in roles.values(): - want = prior == "1" and found.get(r["select_key"]) == r["provider"] - ctx.check(r["active"] is want, "role %s reads active=%s with %s=%r", r["role"], r["active"], - r["select_key"], found.get(r["select_key"])) - try: - if prior != "0": - # the one write sweeps what pointed at the cloud - st, body = fc.post("/settings", data={"cloud_enabled": "0"}) - ctx.check(st == 200, "cloud_enabled=0 -> %s %s", st, body) - ctx.log("cloud mode turned off for the test (found %s)", ev["found"]) - swept = fc.settings() - if found.get("homing_mode") == "gfcloud": - ctx.check(swept.get("homing_mode") == "none", - "cloud_enabled=0 left homing_mode at %r, expected none", swept.get("homing_mode")) - if found.get("controller_mode") == "cloud": - ctx.check(swept.get("controller_mode") == "grbl", - "cloud_enabled=0 left controller_mode at %r, expected grbl", - swept.get("controller_mode")) - ctx.check(fc.settings().get("cloud_enabled") == "0", "cloud_enabled does not read 0") - off = cloud_entry() - ev["extension_off"] = {"enabled": off.get("enabled"), "active": [r["active"] for r in off["roles"]]} - ctx.check(off.get("enabled") is False and not any(r["active"] for r in off["roles"]), - "the list with cloud mode off: %s", ev["extension_off"]) - before = fc.settings() - for key, val, words in (("controller_mode", "cloud", "cloud mode is not enabled on this machine"), - ("homing_mode", "gfcloud", "cloud homing needs cloud mode enabled")): - st, body = fc.post("/settings", data={key: val}) - ev["settings %s=%s" % (key, val)] = st - ctx.log("POST /settings %s=%s -> %s %s", key, val, st, body if isinstance(body, str) else "") - ctx.check(st == 409, "%s=%s with cloud off -> %s, expected 409", key, val, st) - ctx.check(isinstance(body, str) and words in body, "%s=%s was refused in other words: %r", key, val, body) - # the pair in one request is refused too: the request's own switch counts - st, body = fc.post("/settings", data={"cloud_enabled": "0", "controller_mode": "cloud"}) - ctx.check(st == 409, "cloud_enabled=0 with controller_mode=cloud -> %s, expected 409", st) - after = fc.settings() - for key in ("controller_mode", "homing_mode"): - ctx.check(after.get(key) == before.get(key), "%s changed by a refused write: %r -> %r", - key, before.get(key), after.get(key)) - # on again takes the typed phrase, as the cloud step asks for it - st, body = fc.post("/settings", data={"cloud_enabled": "1"}) - ev["settings cloud_enabled=1 no phrase"] = st - ctx.log("POST /settings cloud_enabled=1 (no phrase) -> %s %s", st, body if isinstance(body, str) else "") - ctx.check(st == 400, "cloud_enabled=1 without the phrase -> %s, expected 400", st) - ctx.check(fc.settings().get("cloud_enabled") == "0", "a refused cloud_enabled=1 changed the setting") - st, body = fc.post("/mode", data={"controller": "cloud"}) - text = body if isinstance(body, str) else json.dumps(body) - ev["mode_cloud"] = {"status": st, "body": text} - ctx.log("POST /mode controller=cloud -> %s %s", st, text) - ctx.check(st == 409, "POST /mode controller=cloud with cloud off -> %s, expected 409", st) - ctx.check("cloud mode" in text.lower(), "the refusal does not name cloud mode: %r", text) - m = mode(fc) - ctx.check(m.get("mode") == "grbl" and m.get("controller") == "running", - "the machine did not stay in GRBL mode: %s", m) - finally: - if prior == "": - st, body = fc.post("/settings", params={"cloud_enabled": ""}) - elif prior == "1": - st, body = fc.post("/settings", data={"cloud_enabled": "1", "phrase": "I UNDERSTAND"}) - else: - st, body = fc.post("/settings", data={"cloud_enabled": prior}) - ctx.log("restore cloud_enabled=%r -> %s", prior, st) - # the swept choices go back after cloud mode is on again - for key, cloud_val in (("homing_mode", "gfcloud"), ("controller_mode", "cloud")): - if found.get(key) == cloud_val: - st, body = fc.post("/settings", data={key: cloud_val}) - ctx.log("restore %s=%r -> %s", key, cloud_val, st) - after = fc.settings() - for key in ("cloud_enabled", "homing_mode", "controller_mode"): - ctx.check((after.get(key) or "") == (found.get(key) or ""), "%s not restored: %r, was %r", - key, after.get(key), found.get(key)) - back = cloud_entry() - ctx.check(back.get("enabled") == e.get("enabled") and back.get("roles") == e.get("roles"), - "the list did not come back as found: %s, was %s", back, e) - - -# ----------------------------------------------------------- extensions - -@test("setup.extensions-consent", title="Extensions are turned on over their own advisory", - subsystem="setup", kind="auto", hardware="takeover", est_min=3, - covers=[("forgectrl", "src/main.c"), ("forgectrl", "src/advisories.*"), ("forgectrl", "src/setup.*"), - ("forgectrl", "src/wiz.c"), ("forgectrl", "src/settings.*"), - ("forgectrl", "docs/advisories/extensions.md"), ("forgectrl", "src/ui/embed_docs.cmake")], - requires=["forgectrl.settings-bounds", "setup.advisories-rehash"], - description="The Extensions advisory is on demand: GET /wiz does not list it among the " - "first-run documents, GET /advisories/extensions serves it as markdown with an " - "ETag equal to the SHA-256 of the body, and the first-run accept refuses it (400) " - "and leaves the press. The test turns ext_enabled off itself and puts it back as " - "found. From off, POST /settings ext_enabled=1 is refused without the advisory's " - "hash and with a stale one (409), and without the typed phrase and with the phrase " - "in another case (400); each refusal leaves the setting at 0 and the record on " - "disk untouched. A request that carries the hash and the phrase beside a write the " - "daemon refuses (controller_mode=cloud with cloud_enabled=0, 409) is refused whole " - "and records nothing. With the " - "hash and the phrase it is accepted (200): the setting reads 1, the record on disk " - "holds on_demand.extensions with that hash and the typed method, the first-run " - "documents and their press are as they were, and the data directory has gained " - "the search bit for group and others and nothing else (a package's account walks " - "through it to its own files). Re-sending 1 while it stands and " - "sending 0 ask for nothing. The previous record is put back under a forgectrl " - "restart.") -def extensions_consent(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - raw = read_file(record_path()) - ctx.check(raw, "no setup record at %s", record_path()) - before = wiz(fc) - ev["before"] = wiz_summary(before) - prior = fc.settings().get("ext_enabled") or "" - dir_mode = os.stat(data_dir()).st_mode & 0o7777 - ev["found"] = {"ext_enabled": prior, "data_dir_mode": "%04o" % dir_mode} - - def on_disk(): - try: - return json.loads((read_file(record_path()) or b"{}").decode("utf-8")) - except ValueError: - return {} - - ctx.check("extensions" not in [d.get("id") for d in before.get("documents") or []], - "GET /wiz lists the Extensions advisory among the first-run documents") - st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - ev["etag"] = etag - ctx.log("GET /advisories/extensions -> %s, %d bytes, ETag %s", st, len(body), etag) - ctx.check(st == 200 and body, "GET /advisories/extensions -> %s", st) - ctx.check((hdrs.get("content-type") or "").startswith("text/markdown"), - "the document is not served as markdown: %r", hdrs.get("content-type")) - ctx.check(etag == hashlib.sha256(body).hexdigest(), "the ETag is not the SHA-256 of the body") - - try: - st, reply = fc.post("/wiz/advisories/accept", - data={"doc": "extensions", "hash": etag, "phrase": SAFETY_PHRASE}) - ev["first_run_accept"] = st - ctx.log("the first-run accept of the on-demand document -> %s %s", st, reply) - ctx.check(st == 400, "the first-run accept of the on-demand document -> %s, expected 400", st) - ctx.check(wiz(fc).get("acceptance_done") == before.get("acceptance_done"), - "the refused accept changed the press record") - - if prior == "1": - st, reply = fc.post("/settings", data={"ext_enabled": "0"}) - ctx.check(st == 200, "ext_enabled=0 -> %s %s", st, reply) - ctx.log("extensions turned off for the test (found %r)", prior) - ctx.check((fc.settings().get("ext_enabled") or "0") == "0", "ext_enabled does not read off") - start = on_disk() - - for name, form, want, words in ( - ("no hash", {"ext_enabled": "1"}, 409, "read the Extensions advisory first"), - ("stale hash", {"ext_enabled": "1", "advisory": "0" * 64, "phrase": SAFETY_PHRASE}, 409, - "read the Extensions advisory first"), - ("no phrase", {"ext_enabled": "1", "advisory": etag}, 400, "type I UNDERSTAND"), - ("phrase in another case", {"ext_enabled": "1", "advisory": etag, - "phrase": SAFETY_PHRASE.lower()}, 400, "type I UNDERSTAND"), - ("good consent beside a refused write", {"ext_enabled": "1", "advisory": etag, - "phrase": SAFETY_PHRASE, "cloud_enabled": "0", - "controller_mode": "cloud"}, - 409, "cloud mode is not enabled on this machine")): - st, reply = fc.post("/settings", data=form) - ev[name] = st - ctx.log("POST /settings ext_enabled=1, %s -> %s %s", name, st, reply if isinstance(reply, str) else "") - ctx.check(st == want, "%s -> %s, expected %s", name, st, want) - ctx.check(isinstance(reply, str) and words in reply, "%s was refused in other words: %r", name, reply) - ctx.check((fc.settings().get("ext_enabled") or "0") == "0", "%s: a refused write turned extensions on", name) - ctx.check(on_disk() == start, "%s: a refused write changed the record on disk", name) - - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - ev["accept"] = st - ctx.log("POST /settings ext_enabled=1 with the hash and the phrase -> %s", st) - ctx.check(st == 200, "the consent -> %s %r", st, reply) - ctx.check(fc.settings().get("ext_enabled") == "1", "ext_enabled does not read 1") - rec = on_disk() - entry = (rec.get("on_demand") or {}).get("extensions") or {} - ev["recorded"] = entry - ctx.check(entry.get("hash") == etag and entry.get("method") == "typed" and entry.get("accepted"), - "the record on disk does not carry the acceptance: %s", entry) - ctx.check("extensions" not in (rec.get("advisories") or {}), - "the acceptance landed among the first-run documents") - # a package's account walks through the data directory to its own files: the - # consent adds the search bit, and nothing else - now_mode = os.stat(data_dir()).st_mode & 0o7777 - ev["data_dir_mode_after"] = "%04o" % now_mode - ctx.check(now_mode == dir_mode | 0o011, "the data directory went from %04o to %04o, expected %04o", - dir_mode, now_mode, dir_mode | 0o011) - ctx.check(rec.get("advisories") == start.get("advisories") and rec.get("acceptance") == start.get("acceptance"), - "the first-run documents or their press moved") - after = wiz(fc) - ctx.check(after.get("acceptance_done") == before.get("acceptance_done") and after.get("gate") == before.get("gate"), - "the machine's gate moved with the consent: %s", wiz_summary(after)) - for form in ({"ext_enabled": "1"}, {"ext_enabled": "0"}): - st, reply = fc.post("/settings", data=form) - ctx.check(st == 200, "%s while on -> %s, expected 200", form, st) - ctx.check(fc.settings().get("ext_enabled") == "0", "ext_enabled=0 did not turn extensions off") - finally: - if prior == "": - st, reply = fc.post("/settings", params={"ext_enabled": ""}) - elif prior == "1": - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - else: - st, reply = fc.post("/settings", data={"ext_enabled": prior}) - ctx.log("restore ext_enabled=%r -> %s", prior, st) - if prior != "1": - os.chmod(data_dir(), dir_mode) - with ctx.takeover(): - write_file(record_path(), raw) - ctx.log("the previous record is back under a restart") - - ctx.check((fc.settings().get("ext_enabled") or "") == prior, "ext_enabled not restored: %r, was %r", - fc.settings().get("ext_enabled"), prior) - ctx.check(read_file(record_path()) == raw, "the record on disk is not the one found") - ctx.check(prior == "1" or os.stat(data_dir()).st_mode & 0o7777 == dir_mode, "the data directory's mode is not the one found") - final = wiz(fc) - ev["after_restore"] = wiz_summary(final) - ctx.check(final.get("acceptance_done") == before.get("acceptance_done") and final.get("gate") == before.get("gate"), - "the machine does not read as before: %s", wiz_summary(final)) - - # ---------------------------------------------------------- the operator @test("setup.factory-return", title="The return to the factory firmware is guarded", @@ -1468,16 +972,15 @@ def finished_first_run(ctx): "the controller did not come back after the setup: %s", m) -_FIRST_RUN_REQUIRES = ["setup.gate-blocks-controllers", "setup.advisories-rehash"] - - @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")], - requires=_FIRST_RUN_REQUIRES, actions=["button"], + ("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), " @@ -1612,7 +1115,7 @@ def first_run_flow(ctx): kind="operator", hardware="takeover", est_min=5, covers=[("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/ui/forms.js"), ("forgectrl", "src/ui/md.js"), ("forgectrl", "src/ui/login.*"), ("forgectrl", "src/ui/embed_docs.cmake"), ("forgectrl", "src/tls.*")], - requires=_FIRST_RUN_REQUIRES + ["setup.first-run-flow"], + requires=["setup.first-run-flow"], actions=["button"], hands=["workstation"], steps=["From the workstation open https://
/ and accept the self-signed certificate.", "Click through the setup: scroll each document to its end and accept it (type I " diff --git a/forgetest/tests/test_setup_suite.py b/forgetest/tests/test_setup_suite.py index 171f7b7..eb31510 100644 --- a/forgetest/tests/test_setup_suite.py +++ b/forgetest/tests/test_setup_suite.py @@ -5,9 +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, the settle rule for a -gated supervisor, and the cloud-off surface test driven end to end -against the fake daemon.""" +machine's name, the cookie parsing, the LED cue, and the settle rule +for a gated supervisor.""" import json import os import shutil @@ -16,13 +15,10 @@ import unittest import helpers from forgetest import baseline, catalog -from forgetest.runner import Context, Failed, Run from forgetest.suite import setup -IDS = ("setup.gate-blocks-controllers", - "setup.advisories-rehash", "setup.account-login", "setup.https-only-writes", - "setup.ssh-until-reboot", "setup.cloud-disabled-surface", - "setup.factory-return", "setup.machine-name", "setup.first-run-flow", +IDS = ("setup.account-login", "setup.https-only-writes", + "setup.ssh-until-reboot", "setup.factory-return", "setup.machine-name", "setup.first-run-flow", "setup.first-run-page", "setup.record-export", "setup.mirror") OPERATOR = ("setup.first-run-flow", "setup.first-run-page") @@ -46,9 +42,7 @@ class RegistrationTests(unittest.TestCase): self.assertNotIn("setup.sheet", self.reg) # the two burns, each on its own scrap def test_the_record_swapping_tests_are_takeovers(self): - for tid in ("setup.gate-blocks-controllers", "setup.advisories-rehash", - "setup.first-run-flow", "setup.first-run-page", - "setup.mirror"): + for tid in ("setup.first-run-flow", "setup.first-run-page", "setup.mirror"): self.assertEqual(self.reg[tid].hardware, "takeover", tid) self.assertEqual(self.reg["setup.record-export"].hardware, "api") @@ -75,9 +69,8 @@ class RegistrationTests(unittest.TestCase): self.assertEqual(setup.REQUIRED_STEPS, ("preferences", "machine", "cloud", "cooling.flow")) self.assertIn("cooling.flow", setup.REQUIRED_STEPS) - def test_the_gate_test_starts_from_an_open_gate_and_the_intake_tests_from_a_count(self): + def test_the_intake_tests_start_from_a_count(self): reg = self.reg - self.assertIs(reg["setup.gate-blocks-controllers"].precheck, setup.gate_is_open) from forgetest.suite import setup_dark for tid in ("setup.check-intakes", "setup.check-airflow"): self.assertIs(reg[tid].precheck, setup_dark.intakes_counted, tid) @@ -376,123 +369,5 @@ class SshdPolicyTests(unittest.TestCase): hw.run = real -class CloudDisabledSurfaceTests(unittest.TestCase): - """The cloud-off surface test against the fake daemon: cloud_enabled - is turned off with one write that sweeps the cloud choices, the three - cloud-pointing writes and the phrase-less cloud_enabled=1 are refused, - the settings are left alone, and the prior values come back.""" - - def setUp(self): - self.fake = helpers.FakeForgectrl().start() - self.fake.state["settings"].update({"controller_mode": "grbl", "homing_mode": "none", - "cloud_enabled": "1", "ui_units": "metric"}) - - def on_post(path, form): - s = self.fake.state["settings"] - if path == "/settings": - # as the daemon rules it: on from off takes the typed phrase - if form.get("cloud_enabled") == "1" and s.get("cloud_enabled") != "1" \ - and form.get("phrase") != "I UNDERSTAND": - return 400, "type I UNDERSTAND to turn cloud mode on" - # the request's own cloud choices, as main.c checks them - enabled = form.get("cloud_enabled", s.get("cloud_enabled")) == "1" - if not enabled and form.get("controller_mode") == "cloud": - return 409, "cloud mode is not enabled on this machine" - if not enabled and form.get("homing_mode") == "gfcloud": - return 409, "cloud homing needs cloud mode enabled" - for k, v in form.items(): - if k == "phrase": - continue - if v == "": - s.pop(k, None) - else: - s[k] = v - # off sweeps what pointed at the cloud, as the cloud step does - if form.get("cloud_enabled") == "0": - if "homing_mode" not in form and s.get("homing_mode") == "gfcloud": - s["homing_mode"] = "none" - if "controller_mode" not in form and s.get("controller_mode") == "cloud": - s["controller_mode"] = "grbl" - return 200, dict(s) - if path == "/mode" and form.get("controller") == "cloud" and s.get("cloud_enabled") != "1": - return 409, "cloud mode is not enabled on this machine" - return None - self.fake.on_post = on_post - self.list_lies = None # a test sets this to break the list one way - - def on_get(path, query): - if path != "/extensions": - return None - s = self.fake.state["settings"] - on = s.get("cloud_enabled") == "1" - roles = [{"role": role, "provider": prov, "kind": kind, "select_key": key, "fallback": back, - "active": on and s.get(key) == prov} - for role, prov, kind, key, back in (("homing", "gfcloud", "runner-fd", "homing_mode", "none"), - ("controller", "cloud", "supervised", "controller_mode", - "grbl"))] - if self.list_lies == "stale-active": - roles[0]["active"] = True - if self.list_lies == "always-on": - on = True - return 200, {"extensions": [{"id": "cloud", "name": "Glowforge cloud mode", "builtin": True, - "enabled": on, "enable_key": "cloud_enabled", "setup_step": "cloud", - "tab": "gfcloud", "roles": roles}]} - self.fake.on_get = on_get - - def tearDown(self): - self.fake.stop() - - def run_test(self): - t = catalog.load_suite()["setup.cloud-disabled-surface"] - run = Run("test", t.id, t.title) - t.fn(Context(run, None, t)) - return run - - def test_a_list_that_disagrees_with_the_settings_fails(self): - for lie in ("always-on", "stale-active"): - self.list_lies = lie - self.fake.state["settings"].update({"cloud_enabled": "1", "homing_mode": "gfcloud"}) - with self.assertRaises(Failed, msg=lie): - self.run_test() - - def test_refusals_and_restore(self): - run = self.run_test() - s = self.fake.state["settings"] - self.assertEqual(s["cloud_enabled"], "1") # restored, with the phrase - self.assertEqual(s["controller_mode"], "grbl") - self.assertEqual(s["homing_mode"], "none") - self.assertNotIn("phrase", s) - posts = [(p, f) for p, f in self.fake.posts] - self.assertIn(("/settings", {"cloud_enabled": "0"}), posts) - self.assertIn(("/settings", {"cloud_enabled": "1"}), posts) # refused: no phrase - self.assertIn(("/settings", {"cloud_enabled": "1", "phrase": "I UNDERSTAND"}), posts) - self.assertIn(("/mode", {"controller": "cloud"}), posts) - self.assertEqual(run.evidence["mode_cloud"]["status"], 409) - self.assertEqual(run.evidence["settings cloud_enabled=1 no phrase"], 400) - self.assertEqual(run.evidence["found"]["cloud_enabled"], "1") - - def test_cloud_homing_and_boot_mode_are_swept_by_the_one_write_and_go_back(self): - # As the daemon rules it: cloud_enabled=0 takes the gfcloud homing - # and the cloud boot mode down with it, so the test writes the one - # key, checks the sweep, and puts the three back, cloud mode first. - self.fake.state["settings"].update({"controller_mode": "cloud", "homing_mode": "gfcloud"}) - run = self.run_test() - s = self.fake.state["settings"] - self.assertEqual((s["cloud_enabled"], s["homing_mode"], s["controller_mode"]), ("1", "gfcloud", "cloud")) - posts = [(p, f) for p, f in self.fake.posts if p == "/settings"] - order = [f for _, f in posts] - self.assertEqual(order[:1], [{"cloud_enabled": "0"}]) - self.assertEqual(order[-3:], [{"cloud_enabled": "1", "phrase": "I UNDERSTAND"}, - {"homing_mode": "gfcloud"}, {"controller_mode": "cloud"}]) - self.assertEqual(run.evidence["found"], {"cloud_enabled": "1", "homing_mode": "gfcloud", - "controller_mode": "cloud"}) - - def test_an_unset_value_is_cleared_back(self): - self.fake.state["settings"].pop("cloud_enabled") - self.run_test() - self.assertNotIn("cloud_enabled", self.fake.state["settings"]) - self.assertIn(("/settings", {"cloud_enabled": ""}), self.fake.posts) - - if __name__ == "__main__": unittest.main()