acceptance: helper imports move the fingerprints, and the update test verifies a foreign signature

A test's fingerprint covered its own text and its module's shared text
only, so a judge imported from a sibling suite module (laser.py takes
its motion judges from motion.py) could change without moving the
fingerprints of the tests that call it. The shared text of every sibling
module a module imports now rides along, transitively; unit test.

update.slots-and-signature claimed to refuse a tampered signature but
fed fwup one garbage file. It now makes a throwaway key pair on the
machine, signs a tiny archive, checks that the archive verifies with its
own key and fails against the shipped release key, and asks the update
job to apply it without confirm_unsigned: the job refuses it for its
signature before touching the slot.

forgetest/forgetest/catalog.py

@@ -176,12 +176,38 @@ def module_parts(path):
     return parts
 
 
+def sibling_imports(path, seen=None):
+    """The suite modules a module imports with `from .x import ...`, as
+    paths, transitively, in a stable order. A helper a test calls from
+    a sibling module is part of the judge, so its text belongs in the
+    test's fingerprint."""
+    if seen is None:
+        seen = []
+    with open(path, "rb") as f:
+        text = f.read().replace(b"\r\n", b"\n").decode("utf-8")
+    try:
+        tree = ast.parse(text)
+    except SyntaxError:
+        return seen
+    here = os.path.dirname(path)
+    for node in tree.body:
+        if isinstance(node, ast.ImportFrom) and node.level == 1 and node.module:
+            sib = os.path.join(here, node.module.replace(".", os.sep) + ".py")
+            if os.path.isfile(sib) and sib not in seen and sib != path:
+                seen.append(sib)
+                sibling_imports(sib, seen)
+    return seen
+
+
 def implementation_sha(path, test_id):
-    """The implementation hash of one test (see Test.source_sha)."""
+    """The implementation hash of one test (see Test.source_sha): the
+    test's own text, the shared text of its module, and the shared text
+    of every sibling suite module the module imports."""
     shared, own = module_parts(path)
     if test_id not in own:
         return source_file_sha(path)
-    return _manifest.sha256_text("%s:%s" % (shared, own[test_id]))
+    helpers = [module_parts(p)[0] for p in sorted(sibling_imports(path))]
+    return _manifest.sha256_text("%s:%s" % (":".join([shared] + helpers), own[test_id]))
 
 
 def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers=(),

forgetest/forgetest/suite/update.py

@@ -1,6 +1,8 @@
 """update.* - the A/B slot inventory and the firmware verification path."""
 import os
 import tempfile
+import time
+import shutil
 
 from ..catalog import test
 from .. import hw
@@ -9,12 +11,14 @@ _UPDATE_COVERS = [("forgectrl", "src/update.c"), ("forgectrl", "src/update.h"),
                   ("ffboot", "**")]
 
 
-@test("update.slots-and-signature", title="Boot slots readable, unsigned/tampered archives refused",
-      subsystem="update", kind="auto", est_min=1,
+@test("update.slots-and-signature", title="Boot slots readable, unsigned and foreign-signed archives refused",
+      subsystem="update", kind="auto", est_min=2,
       covers=_UPDATE_COVERS, requires=["forgectrl.auth"],
       description="/slots reports the A/B inventory consistent with `ffboot -l`; /update/status "
-                  "answers; `fwup` refuses a garbage archive and a tampered signature against the "
-                  "shipped release key. Nothing is written to any slot.")
+                  "answers; `fwup` refuses a garbage archive. A tiny archive signed with a "
+                  "throwaway key made on the spot verifies with its own key and fails against the "
+                  "shipped release key, and an apply of it without confirm_unsigned is refused by "
+                  "the update job before anything is written. Nothing is written to any slot.")
 def slots_and_signature(ctx):
     fc = ctx.forgectrl
     ev = ctx.evidence
@@ -46,3 +50,69 @@ def slots_and_signature(ctx):
         ctx.check(rc != 0, "fwup accepted a garbage archive")
     finally:
         os.unlink(garbage)
+
+    # A foreign signature: a throwaway key pair signs a tiny archive. The
+    # shipped key must refuse it, its own key must accept it, and the
+    # apply job must refuse it without confirm_unsigned, before it
+    # touches the slot.
+    work = tempfile.mkdtemp(prefix="forgetest-fw-")
+    staged = "/data/forgefirm/upload.fw"
+    try:
+        with open(os.path.join(work, "note.txt"), "w") as f:
+            f.write("forgetest foreign-signature drill\n")
+        with open(os.path.join(work, "fwup.conf"), "w") as f:
+            f.write('meta-product = "forgetest"\nmeta-version = "0.0.0-test"\n'
+                    'file-resource note.txt { host-path = "note.txt" }\n'
+                    'task complete { on-resource note.txt { raw_write(0) } }\n')
+        rc, out = hw.run(["sh", "-c", "cd %s && fwup -g" % work], timeout=60)
+        ev["fwup_gen_rc"] = rc
+        ctx.check(rc == 0 and os.path.exists(os.path.join(work, "fwup-key.pub")),
+                  "fwup -g did not make a key pair (rc %s): %s", rc, out.strip()[:200])
+        plain, signed = os.path.join(work, "plain.fw"), os.path.join(work, "signed.fw")
+        rc, out = hw.run(["sh", "-c", "cd %s && fwup -c -f fwup.conf -o plain.fw && "
+                          "fwup -S -s fwup-key.priv -i plain.fw -o signed.fw" % work], timeout=60)
+        ctx.check(rc == 0 and os.path.exists(signed), "could not make the signed archive (rc %s): %s",
+                  rc, out.strip()[:200])
+        rc_own, _ = hw.run(["fwup", "-V", "-i", signed, "-p", os.path.join(work, "fwup-key.pub")], timeout=30)
+        rc_ship, out = hw.run(["fwup", "-V", "-i", signed, "-p", key], timeout=30)
+        ev["fwup_foreign"] = {"own_key_rc": rc_own, "shipped_key_rc": rc_ship}
+        ctx.log("fwup -V signed: own key rc %s, shipped key rc %s", rc_own, rc_ship)
+        ctx.check(rc_own == 0, "the archive does not verify with the key that signed it")
+        ctx.check(rc_ship != 0, "the shipped release key accepted a foreign signature")
+
+        # The apply path. The target is the slot not booted; a slot already
+        # selected for the next boot is refused for its own reason, which
+        # this drill records and steps over.
+        target = None
+        for name, si in (slots.get("slots") or {}).items():
+            if name in ("a", "b") and isinstance(si, dict) and not si.get("booted"):
+                target = name
+        ctx.check(target is not None, "no inactive firmware slot in /slots: %s", slots)
+        shutil.copyfile(signed, staged)
+        st, body = fc.post("/update/apply", params={"slot": target, "file": "upload"})
+        ev["apply"] = {"status": st, "body": body}
+        ctx.log("POST /update/apply slot=%s file=upload (no confirm_unsigned) -> %s %s", target, st, body)
+        if st == 409 and isinstance(body, dict) and "next boot" in str(body.get("error", "")):
+            ctx.log("slot %s is selected for the next boot: the apply is refused before the "
+                    "signature check, which is its own guard", target)
+        else:
+            ctx.check(st == 200, "the apply job did not start: %s %s", st, body)
+            result = None
+            t0 = time.time()
+            while time.time() - t0 < 60:
+                ctx.sleep(1)
+                st, us = fc.get("/update/status")
+                if isinstance(us, dict) and not us.get("running"):
+                    result = us.get("result")
+                    break
+            ev["apply_result"] = result
+            ctx.log("apply result: %s", result)
+            ctx.check(isinstance(result, dict) and result.get("ok") is False
+                      and "not signed" in str(result.get("error", "")),
+                      "the apply of a foreign-signed archive was not refused for its signature: %s", result)
+    finally:
+        try:
+            os.unlink(staged)
+        except OSError:
+            pass
+        shutil.rmtree(work, ignore_errors=True)

forgetest/tests/test_manifest.py

@@ -242,6 +242,28 @@ def two(ctx):
         self.assertNotEqual(a1, a2)
         self.assertNotEqual(b1, b2)
 
+    def test_a_helper_edit_in_an_imported_sibling_moves_the_tests(self):
+        # laser.py imports its judges from motion.py: an edit to such a
+        # helper changes the judge, so it has to move the fingerprints of
+        # the tests that import it, transitively.
+        sib = os.path.join(self.tmp, "judge.py")
+        with open(sib, "w", newline="\n") as f:
+            f.write("def judge(x):\n    return x > 1\n")
+        catalog._PARTS.pop(sib, None)
+        self.write("from .judge import judge\n" + self.MODULE)
+        a1, b1 = self.shas()
+        with open(sib, "w", newline="\n") as f:
+            f.write("def judge(x):\n    return x > 2\n")
+        catalog._PARTS.pop(sib, None)
+        a2, b2 = self.shas()
+        self.assertNotEqual(a1, a2)
+        self.assertNotEqual(b1, b2)
+        # a module that is not imported changes nothing
+        other = os.path.join(self.tmp, "other.py")
+        with open(other, "w", newline="\n") as f:
+            f.write("def o():\n    return 1\n")
+        self.assertEqual((a2, b2), self.shas())
+
     def test_line_endings_do_not_count(self):
         a1, b1 = self.shas()
         catalog._PARTS.pop(self.path, None)