diff --git a/forgetest/forgetest/catalog.py b/forgetest/forgetest/catalog.py index 7374e7e..27dd283 100644 --- a/forgetest/forgetest/catalog.py +++ b/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=(), diff --git a/forgetest/forgetest/suite/update.py b/forgetest/forgetest/suite/update.py index 8e327c6..312fed3 100644 --- a/forgetest/forgetest/suite/update.py +++ b/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) diff --git a/forgetest/tests/test_manifest.py b/forgetest/tests/test_manifest.py index 6e1b655..42c4739 100644 --- a/forgetest/tests/test_manifest.py +++ b/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)