exthost.panel-install: the key goes in through the panel, and the box holds the button

The test added the owner's key by copying the file. It now adds it the way
an operator does: POST /ext/key with the machine's button held. Without the
button it is 409 and no file lands; a name with a space, a name that is a
path, and no key at all are 400; a key that is no key is 409 from the host;
with the button held it is added and listed with its id. Removed again, the
same archive reads unverified, and removing a key that is not there is 409.

The bench actuator's press was 200 ms, the firmware's default, and a
request that must reach the machine while the button is down often missed
it: the unverified install took three presses on one run and all ten on
another. ctx.act() now takes `ms`, the fixture client passes it to the box
(the firmware clamps it into 20 to 500), and the two button-held steps ask
for 500. Both landed on the first or second press afterward.

Proven. The unit suite: 451 tests, 0 undefined names (both stub fixtures
take the new argument, and the fixture test pins that the box is asked for
the longest press). On the bench reference, image 20260921161446 with the
cross-built forgectrl and extension host mounted over the image's:
exthost.panel-install PASS, exthost.package-routes PASS,
exthost.hold-pause-tier PASS. Against the image's own daemons the install
test FAILS, as it should.

Acceptance. exthost.panel-install covers forgectrl's src/extpkg.*,
src/main.c, src/auth.* and forgeext's src/main.c, src/install.*,
src/pkg.*; the runner and the fixture client are harness, outside the
suite and outside every fingerprint.

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):
+    def _press(self, ms=None):
-        """One press, spaced from the last one. The controller detects
+        """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

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:

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
+        # the same key, now the owner's, added through the panel with the machine's button held
-        shutil.copy(pub, owner_key)
+        with open(pub) as f:
-        os.chmod(owner_key, 0o644)
+            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")

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"):

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):
+            def act(self, channel, state, ms=None):
-                self.done.append((channel, state))
+                self.done.append((channel, state) if ms is None else (channel, state, ms))
                 self_outer.lid(state != "open")
         self_outer = self