diff --git a/forgetest/forgetest/fixture.py b/forgetest/forgetest/fixture.py index e57aa91..aeed646 100644 --- a/forgetest/forgetest/fixture.py +++ b/forgetest/forgetest/fixture.py @@ -297,11 +297,11 @@ class Fixture: return bool(st and st.get("button_enabled")) return True - def act(self, channel, state): + def act(self, channel, state, ms=None): if channel == "button": if state != "press": raise FixtureError("the button is only ever pressed") - st, body = self._press() + st, body = self._press(ms) else: if state not in ("open", "close"): raise FixtureError("%s: unknown state %r" % (channel, state)) @@ -312,8 +312,11 @@ class Fixture: self.last_state = body return body - def _press(self): - """One press, spaced from the last one. The controller detects + def _press(self, ms=None): + """One press, spaced from the last one. `ms` is how long the box + holds it (the firmware clamps it into 20 to 500, and takes 200 when + it is not said): a check that must reach the machine *while* the + button is down asks for the longest press the box can make. The controller detects the button's rising edge, so a press must follow a release it has seen: the next press waits for the last pulse to end (the fixture's own pulse_ms) plus BUTTON_GAP_S. A 409 for a pulse in @@ -322,7 +325,7 @@ class Fixture: wait = self._pulse_end + BUTTON_GAP_S - time.time() if wait > 0: time.sleep(wait) - st, body = self._request("POST", "/button", {}) + st, body = self._request("POST", "/button", {"ms": int(ms)} if ms else {}) if st == 409 and "in progress" in str(body.get("error", "")): deadline = time.time() + PULSE_WAIT_S while time.time() < deadline: @@ -331,7 +334,7 @@ class Fixture: if s == 200 and not b.get("button_pulsing"): break time.sleep(BUTTON_GAP_S) - st, body = self._request("POST", "/button", {}) + st, body = self._request("POST", "/button", {"ms": int(ms)} if ms else {}) if st == 200: self._pulse_end = time.time() + float(body.get("pulse_ms") or 500) / 1000.0 return st, body diff --git a/forgetest/forgetest/runner.py b/forgetest/forgetest/runner.py index c6155ec..a4f85cd 100644 --- a/forgetest/forgetest/runner.py +++ b/forgetest/forgetest/runner.py @@ -366,7 +366,7 @@ class Context: """One of forgectrl's switch readings (lid, interlock_ok, ...).""" return (self.forgectrl.status().get("switches") or {}).get(name) - def act(self, channel, state, until=None, timeout=ACT_TIMEOUT_S, text="", fail=True): + def act(self, channel, state, until=None, timeout=ACT_TIMEOUT_S, text="", fail=True, ms=None): """A machine action by name: ("lid", "open"|"close"), ("interlock", "open"|"close"), ("button", "press"). The bench's actuator performs it when one covers the channel (the runner's @@ -405,7 +405,7 @@ class Context: self.log("ACT %s %s (fixture)", channel, state) rec["by"] = "fixture" try: - fixture.act(channel, state) + fixture.act(channel, state, ms) except _fixture.FixtureError as e: rec["fixture_error"] = str(e) if self.run.unattended: diff --git a/forgetest/forgetest/suite/exthost.py b/forgetest/forgetest/suite/exthost.py index 1b70099..1df9ca1 100644 --- a/forgetest/forgetest/suite/exthost.py +++ b/forgetest/forgetest/suite/exthost.py @@ -1385,7 +1385,10 @@ EXT_STAGE = "/data/forgefirm/tmp/ext-upload.ffx" "the same key added as the owner's, the upload reads community with consent typed: the " "install without the phrase is 400, with the phrase and without the grant 409 in the " "host's words, with a grant that has not the form of a capability 400, and with the phrase " - "and the grant it installs. An install with nothing staged is 409, and a discarded upload " + "and the grant it installs. The key itself goes in through the panel: without the button " + "held it is 409 and no file lands, a name with a space or a path is 400, a key that is no " + "key is 409 from the host, and with the button held it is added and listed with its id; " + "removed again, the same archive reads unverified. An install with nothing staged is 409, and a discarded upload " "is gone. Both packages are removed through the route; the key, the work directory, and " "the staged file are removed and the extension root is as found.") def panel_install(ctx): @@ -1453,7 +1456,7 @@ def panel_install(ctx): # until a request has landed inside one. for _ in range(10): if ctx.act("button", "press", text="HOLD the button now, for a few seconds: an unsigned package is being installed.", - until=held_install, timeout=4, fail=False) is not None: + until=held_install, timeout=4, fail=False, ms=500) is not None: break pkg = installed() ev["unverified_installed"] = {k: (pkg or {}).get(k) for k in ("tier", "grants", "hold", "enabled")} @@ -1463,9 +1466,39 @@ def panel_install(ctx): st, reply = fc.post("/ext/package", data={"id": REF_ID, "action": "remove"}) ctx.check(st == 200 and not installed(), "remove through the route -> %s", st) - # the same key, now the owner's: the typed phrase - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) + # the same key, now the owner's, added through the panel with the machine's button held + with open(pub) as f: + keytext = f.read().strip() + st, reply = fc.post("/ext/key", data={"name": REF_KEY, "key": keytext}) + ev["key_without_the_button"] = [st, reply if isinstance(reply, str) else ""] + ctx.check(st == 409 and isinstance(reply, str) and "button" in reply and not os.path.exists(owner_key), + "a key added without the button held -> %s %r", st, reply) + for name, form, want in (("a name with a space", {"name": "a maker", "key": keytext}, 400), + ("a name that is a path", {"name": "../../etc/passwd", "key": keytext}, 400), + ("no key at all", {"name": REF_KEY}, 400), + ("a key that is no key", {"name": REF_KEY, "key": "this is no key"}, 409)): + st, reply = fc.post("/ext/key", data=form) + ev[name] = st + ctx.check(st == want and not os.path.exists(owner_key), "%s -> %s, expected %s", name, st, want) + keyed = {} + + def held_key(): + st_, reply_ = fc.post("/ext/key", data={"name": REF_KEY, "key": keytext}) + keyed["last"] = [st_, reply_ if isinstance(reply_, str) else "ok"] + return st_ == 200 + + for _ in range(10): + if ctx.act("button", "press", text="HOLD the button now, for a few seconds: a key is being added.", + until=held_key, timeout=4, fail=False, ms=500) is not None: + break + st, doc = fc.get("/ext/status") + keys = {k.get("name"): k.get("key") for k in ((doc or {}).get("keys") or [])} + ev["keys"] = keys + ctx.log("with the button held: %s, the keys now %s", keyed.get("last"), sorted(keys)) + ctx.check(REF_KEY in keys and os.path.exists(owner_key), "the key was not added with the button held: %s (%s)", + sorted(keys), keyed.get("last")) + ctx.check(len(keys.get(REF_KEY) or "") == 64, "the key is listed without its id: %s", keys) + st, doc = upload(archive) ev["community_upload"] = {k: (doc or {}).get(k) for k in ("tier", "consent", "needs_grant")} if isinstance(doc, dict) else doc ctx.check(st == 200 and isinstance(doc, dict) and doc.get("tier") == "community" and doc.get("consent") == "typed", @@ -1488,8 +1521,16 @@ def panel_install(ctx): st, reply = fc.post("/ext/package", data={"id": REF_ID, "action": "remove"}) ctx.check(st == 200 and not installed(), "remove through the route -> %s", st) + # the key removed: the same archive reads unverified again + st, reply = fc.post("/ext/key/remove", data={"name": REF_KEY}) + ctx.check(st == 200 and not os.path.exists(owner_key), "the key was not removed -> %s %r", st, reply) + st, reply = fc.post("/ext/key/remove", data={"name": REF_KEY}) + ctx.check(st == 409, "removing a key that is not there -> %s", st) + st, doc = upload(archive) - ctx.check(st == 200 and os.path.exists(EXT_STAGE), "the third upload -> %s", st) + ev["after_the_key_went"] = (doc or {}).get("tier") if isinstance(doc, dict) else doc + ctx.check(st == 200 and isinstance(doc, dict) and doc.get("tier") == "unverified" and os.path.exists(EXT_STAGE), + "without the key the same archive reads %s", ev["after_the_key_went"]) st, reply = fc.post("/ext/upload/discard") ctx.check(st == 200 and not os.path.exists(EXT_STAGE), "discard -> %s, staged file %s", st, "kept" if os.path.exists(EXT_STAGE) else "gone") diff --git a/forgetest/tests/test_fixture.py b/forgetest/tests/test_fixture.py index d0ce37a..bb38a04 100644 --- a/forgetest/tests/test_fixture.py +++ b/forgetest/tests/test_fixture.py @@ -207,6 +207,10 @@ class ClientTests(unittest.TestCase): f.act("interlock", "open") f.act("button", "press") self.assertEqual(self.dev.calls[-1], ("POST", "/button", {})) + time.sleep(0.15) + # a check that must reach the machine while the button is down asks for the longest press + f.act("button", "press", ms=500) + self.assertEqual(self.dev.calls[-1], ("POST", "/button", {"ms": 500})) time.sleep(0.15) # the fake's pulse ends self.assertEqual(sorted(fx.Fixture.energized(f.status())), ["interlock", "lid"]) f.release() @@ -330,10 +334,10 @@ class StubFixture: def covers(self, channel): return channel in self.channels and (channel != "button" or self.button_enabled) - def act(self, channel, state): + def act(self, channel, state, ms=None): if self.fail: raise fx.FixtureError("the box is off") - self.acts.append((channel, state)) + self.acts.append((channel, state) if ms is None else (channel, state, ms)) if state == "open": self.held.append(channel) if self.fc is not None and channel in ("lid", "interlock"): diff --git a/forgetest/tests/test_operator.py b/forgetest/tests/test_operator.py index 24805b5..0d763a2 100644 --- a/forgetest/tests/test_operator.py +++ b/forgetest/tests/test_operator.py @@ -141,8 +141,8 @@ class ActTests(unittest.TestCase): def covers(self, channel): return channel == "lid" - def act(self, channel, state): - self.done.append((channel, state)) + def act(self, channel, state, ms=None): + self.done.append((channel, state) if ms is None else (channel, state, ms)) self_outer.lid(state != "open") self_outer = self