From bf693454dc689673dc6352041dff801e17bbdc04 Mon Sep 17 00:00:00 2001 From: gspark-etri Date: Fri, 30 Oct 2026 21:09:12 +0900 Subject: [PATCH 1/2] Reject incomplete device-binding verification coverage --- .github/workflows/ci.yml | 2 + conformance/binding/conformance.py | 48 ++++++-- conformance/binding/test_conformance.py | 154 ++++++++++++++++++++++++ 3 files changed, 194 insertions(+), 10 deletions(-) create mode 100644 conformance/binding/test_conformance.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1402977..f2d45e0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,8 @@ jobs: run: test -z "$(gofmt -l .)" || { gofmt -l .; exit 1; } - run: go vet ./... - run: go test -race ./... + - name: device binding verification regressions + run: python3 -m unittest discover -s conformance/binding -p 'test_*.py' -v - name: device binding conformance kit (reference fake binding, B1 and B2) run: python3 conformance/binding/conformance.py --level B2 && python3 conformance/binding/conformance.py --level B1 - name: chunker benchmark (reported, not gated yet) diff --git a/conformance/binding/conformance.py b/conformance/binding/conformance.py index 8c9f01c..977386f 100644 --- a/conformance/binding/conformance.py +++ b/conformance/binding/conformance.py @@ -10,6 +10,7 @@ import hashlib import json import os +import re import subprocess import sys import tempfile @@ -61,11 +62,41 @@ def covers(update, switch): return True -def verify_units(ver, files, what): - ok = ver.get("schema") == "onggi.verification.v1" and ver.get("ok") is True - for u in ver.get("units", []): - want = files.get(u["file"], b"")[u["offset"]:u["offset"] + u["length"]] - ok = ok and len(want) == u["length"] and u["sha256"] == sha(want) +def verify_units(ver, files, what, update, op): + """Require a read-back digest for exactly every planned/restored artifact interval.""" + mode = update["mode"] + manifest = update["manifest" if op == "apply" else "base_manifest"] + if not (isinstance(ver, dict) and ver.get("schema") == "onggi.verification.v1" + and ver.get("ok") is True and ver.get("op") == op + and mode in ("units", "reload") and ver.get("mode") == mode + and isinstance(manifest, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", manifest) + and ver.get("manifest") == manifest and isinstance(ver.get("units"), list)): + return check(False, what) + + def interval(unit): + if not isinstance(unit, dict): + return None + path, off, length = unit.get("file"), unit.get("offset"), unit.get("length") + if (not isinstance(path, str) or path not in files + or type(off) is not int or type(length) is not int + or off < 0 or length <= 0 or off + length > len(files[path])): + return None + return path, off, length + + # Reload rollback restores the base files, whose paths and sizes may differ from the target. + required = update["units"] if mode == "units" else [ + {"file": path, "offset": 0, "length": len(data)} for path, data in files.items()] + expected = [interval(u) for u in required] + actual = [interval(u) for u in ver["units"]] + ok = (None not in expected and None not in actual + and len(set(expected)) == len(expected) and len(set(actual)) == len(actual) + and set(actual) == set(expected)) + if ok: + for unit, (path, off, length) in zip(ver["units"], actual): + ok = ok and unit.get("sha256") == sha(files[path][off:off + length]) + if "bytes_written" in ver: + ok = (ok and type(ver["bytes_written"]) is int + and ver["bytes_written"] == sum(u[2] for u in actual)) return check(ok, what) @@ -121,15 +152,12 @@ def main(argv): if level == "B1": continue _, ver = run(cmd, "apply", "--target", tdir, stdin=up1) - verify_units(ver, target, "apply verifies: unit digests equal the target bytes") - if up1["mode"] == "reload": - check(sorted(u["file"] for u in ver["units"]) == sorted(target), "reload writes every target file") + verify_units(ver, target, "apply verifies: every planned unit matches the target bytes", up1, "apply") _, res3 = run(cmd, "resident") check(res3.get("manifest") == sp["manifest"], "after apply the target is resident") _, rb = run(cmd, "rollback") - check(rb.get("op") == "rollback", "rollback returns a verification") - verify_units(rb, base, "rollback restores: unit digests equal the base bytes") + verify_units(rb, base, "rollback restores: every planned unit matches the base bytes", up1, "rollback") _, res4 = run(cmd, "resident") check(res4.get("manifest") == sp["base_manifest"], "after rollback the base is resident") rc, _ = run(cmd, "rollback", expect_ok=False) diff --git a/conformance/binding/test_conformance.py b/conformance/binding/test_conformance.py new file mode 100644 index 0000000..72be53c --- /dev/null +++ b/conformance/binding/test_conformance.py @@ -0,0 +1,154 @@ +# Copyright 2026 The onggi authors. SPDX-License-Identifier: Apache-2.0 +"""Regression checks for the binding kit's verification boundary (stdlib only).""" +import contextlib +import copy +import hashlib +import io +import os +import tempfile +import unittest +from unittest import mock + +import conformance + + +class VerificationTests(unittest.TestCase): + def setUp(self): + self.old_fails = conformance.FAILS[:] + conformance.FAILS[:] = [] + + def tearDown(self): + conformance.FAILS[:] = self.old_fails + + def fixtures(self, mode="units", op="apply"): + files = {"model.bin": b"abcdef", "config.json": b"cfg"} + plan = {"mode": mode, "manifest": "sha256:" + "1" * 64, + "base_manifest": "sha256:" + "0" * 64, + "units": [{"file": "model.bin", "offset": 1, "length": 3}, + {"file": "config.json", "offset": 0, "length": 3}] if mode == "units" else []} + units = [dict(u, sha256=hashlib.sha256(files[u["file"]][u["offset"]:u["offset"] + u["length"]]).hexdigest()) + for u in plan["units"]] + ver = {"schema": "onggi.verification.v1", "op": op, "mode": mode, + "manifest": plan["manifest" if op == "apply" else "base_manifest"], + "ok": True, "units": units} + return files, plan, ver + + def verify(self, ver, files, plan, op="apply"): + with contextlib.redirect_stdout(io.StringIO()): + return conformance.verify_units(ver, files, "verification", plan, op) + + def test_accepts_exact_planned_units_in_any_order(self): + files, plan, ver = self.fixtures() + ver["units"].reverse() + self.assertTrue(self.verify(ver, files, plan)) + + def test_rejects_missing_duplicate_and_unexpected_intervals(self): + files, plan, ver = self.fixtures() + units = ver["units"] + extra = {"file": "model.bin", "offset": 0, "length": 1, + "sha256": hashlib.sha256(b"a").hexdigest()} + for returned in [[], units[:1], units + units[:1], units + [extra], + [dict(units[0], offset=0, sha256=hashlib.sha256(b"abc").hexdigest()), units[1]]]: + with self.subTest(units=returned): + self.assertFalse(self.verify(dict(ver, units=returned), files, plan)) + + def test_rejects_invalid_unit_fields_without_crashing(self): + files, plan, ver = self.fixtures() + bad = [{"offset": v} for v in [-1, True, 1.0, "1", None, 6]] + bad += [{"length": v} for v in [0, -1, True, 3.0, "3", None, 99]] + bad += [{"file": v} for v in ["unknown", None, [], 1]] + bad += [{"sha256": v} for v in [None, "", "0" * 64, "Z" * 64]] + for changes in bad: + with self.subTest(changes=changes): + units = [dict(ver["units"][0], **changes), ver["units"][1]] + self.assertFalse(self.verify(dict(ver, units=units), files, plan)) + for units in [None, {}, [None], [{}]]: + with self.subTest(units=units): + self.assertFalse(self.verify(dict(ver, units=units), files, plan)) + for key in ["file", "offset", "length", "sha256"]: + broken = copy.deepcopy(ver) + del broken["units"][0][key] + self.assertFalse(self.verify(broken, files, plan)) + + def test_rejects_wrong_or_missing_verification_identity(self): + files, plan, ver = self.fixtures() + for key, wrong in [("schema", "unknown"), ("op", "rollback"), ("mode", "reload"), + ("manifest", plan["base_manifest"]), ("ok", 1), ("ok", False)]: + with self.subTest(key=key, value=wrong): + self.assertFalse(self.verify(dict(ver, **{key: wrong}), files, plan)) + for key in ver: + missing = dict(ver) + del missing[key] + self.assertFalse(self.verify(missing, files, plan)) + for value in [None, [], True]: + self.assertFalse(self.verify(value, files, plan)) + + def test_checks_optional_bytes_written_when_present(self): + files, plan, ver = self.fixtures() + self.assertTrue(self.verify(dict(ver, bytes_written=6), files, plan)) + for value in [0, -1, True, 6.0, "6", None]: + with self.subTest(bytes_written=value): + self.assertFalse(self.verify(dict(ver, bytes_written=value), files, plan)) + + def test_identical_switch_accepts_explicit_empty_units(self): + files, plan, ver = self.fixtures() + plan["units"] = [] + self.assertTrue(self.verify(dict(ver, units=[], bytes_written=0), files, plan)) + self.assertFalse(self.verify(ver, files, plan)) + missing = dict(ver) + del missing["units"] + self.assertFalse(self.verify(missing, files, plan)) + + def test_reload_requires_every_whole_target_file(self): + files, plan, ver = self.fixtures(mode="reload") + units = [{"file": "model.bin", "offset": 0, "length": 6, + "sha256": hashlib.sha256(b"abcdef").hexdigest()}, + {"file": "config.json", "offset": 0, "length": 3, + "sha256": hashlib.sha256(b"cfg").hexdigest()}] + self.assertTrue(self.verify(dict(ver, units=units), files, plan)) + for returned in [[], units[:1], units + units[:1], + [dict(units[0], length=3, sha256=hashlib.sha256(b"abc").hexdigest()), units[1]]]: + with self.subTest(units=returned): + self.assertFalse(self.verify(dict(ver, units=returned), files, plan)) + + def test_reload_rollback_uses_base_files_and_lengths(self): + _, plan, ver = self.fixtures(mode="reload", op="rollback") + base = {"model.bin": b"ab", "config.json": b"longer config", "base-only": b"old"} + units = [{"file": name, "offset": 0, "length": len(data), + "sha256": hashlib.sha256(data).hexdigest()} for name, data in base.items()] + self.assertTrue(self.verify(dict(ver, units=units), base, plan, "rollback")) + self.assertFalse(self.verify(dict(ver, units=units[:2]), base, plan, "rollback")) + self.assertFalse(self.verify(dict(ver, units=units, manifest=plan["manifest"]), base, plan, "rollback")) + + def test_units_rollback_requires_all_planned_base_intervals(self): + _, plan, ver = self.fixtures(op="rollback") + base = {"model.bin": b"012345", "config.json": b"old"} + units = [{"file": "model.bin", "offset": 1, "length": 3, + "sha256": hashlib.sha256(b"123").hexdigest()}, + {"file": "config.json", "offset": 0, "length": 3, + "sha256": hashlib.sha256(b"old").hexdigest()}] + self.assertTrue(self.verify(dict(ver, units=units), base, plan, "rollback")) + self.assertFalse(self.verify(dict(ver, units=units[:1]), base, plan, "rollback")) + self.assertFalse(self.verify(dict(ver, units=units, op="apply"), base, plan, "rollback")) + + def test_rejects_empty_apply_verification_for_planned_units(self): + real_run = conformance.run + + def omit_units(cmd, op, *args, **kwargs): + rc, doc = real_run(cmd, op, *args, **kwargs) + if op == "apply": + doc["units"] = [] + return rc, doc + + with tempfile.TemporaryDirectory() as work, \ + mock.patch.dict(os.environ), \ + mock.patch.object(conformance.tempfile, "mkdtemp", return_value=work), \ + mock.patch.object(conformance.golden_data, "CASES", ("inplace",)), \ + mock.patch.object(conformance, "run", side_effect=omit_units), \ + contextlib.redirect_stdout(io.StringIO()): + os.environ.pop("ONGGI_FAKE_LEVEL", None) + self.assertEqual(conformance.main(["--level", "B2"]), 1) + + +if __name__ == "__main__": + unittest.main() From 39696e77b62d6095e3667164d2bcc9909f680d91 Mon Sep 17 00:00:00 2001 From: gspark-etri Date: Fri, 30 Oct 2026 21:16:40 +0900 Subject: [PATCH 2/2] Preserve empty reload files and device byte accounting semantics --- conformance/binding/conformance.py | 7 ++++--- conformance/binding/test_conformance.py | 26 ++++++++++++++++++++++--- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/conformance/binding/conformance.py b/conformance/binding/conformance.py index 977386f..4335a3a 100644 --- a/conformance/binding/conformance.py +++ b/conformance/binding/conformance.py @@ -63,7 +63,7 @@ def covers(update, switch): def verify_units(ver, files, what, update, op): - """Require a read-back digest for exactly every planned/restored artifact interval.""" + """Verify every planned/restored interval; bytes_written is a nonnegative device counter.""" mode = update["mode"] manifest = update["manifest" if op == "apply" else "base_manifest"] if not (isinstance(ver, dict) and ver.get("schema") == "onggi.verification.v1" @@ -79,7 +79,8 @@ def interval(unit): path, off, length = unit.get("file"), unit.get("offset"), unit.get("length") if (not isinstance(path, str) or path not in files or type(off) is not int or type(length) is not int - or off < 0 or length <= 0 or off + length > len(files[path])): + or off < 0 or length < 0 or off + length > len(files[path]) + or (length == 0 and not (mode == "reload" and off == 0 and len(files[path]) == 0))): return None return path, off, length @@ -96,7 +97,7 @@ def interval(unit): ok = ok and unit.get("sha256") == sha(files[path][off:off + length]) if "bytes_written" in ver: ok = (ok and type(ver["bytes_written"]) is int - and ver["bytes_written"] == sum(u[2] for u in actual)) + and ver["bytes_written"] >= 0) return check(ok, what) diff --git a/conformance/binding/test_conformance.py b/conformance/binding/test_conformance.py index 72be53c..b50fc40 100644 --- a/conformance/binding/test_conformance.py +++ b/conformance/binding/test_conformance.py @@ -1,5 +1,5 @@ # Copyright 2026 The onggi authors. SPDX-License-Identifier: Apache-2.0 -"""Regression checks for the binding kit's verification boundary (stdlib only).""" +"""Check artifact coverage and schema-valid device write counters (stdlib only).""" import contextlib import copy import hashlib @@ -85,8 +85,10 @@ def test_rejects_wrong_or_missing_verification_identity(self): def test_checks_optional_bytes_written_when_present(self): files, plan, ver = self.fixtures() - self.assertTrue(self.verify(dict(ver, bytes_written=6), files, plan)) - for value in [0, -1, True, 6.0, "6", None]: + for value in [6, 0, 8]: + with self.subTest(bytes_written=value): + self.assertTrue(self.verify(dict(ver, bytes_written=value), files, plan)) + for value in [-1, True, 6.0, "6", None]: with self.subTest(bytes_written=value): self.assertFalse(self.verify(dict(ver, bytes_written=value), files, plan)) @@ -120,6 +122,24 @@ def test_reload_rollback_uses_base_files_and_lengths(self): self.assertFalse(self.verify(dict(ver, units=units[:2]), base, plan, "rollback")) self.assertFalse(self.verify(dict(ver, units=units, manifest=plan["manifest"]), base, plan, "rollback")) + def verify_reload_empty_file(self, op): + _, plan, ver = self.fixtures(mode="reload", op=op) + files = {"empty.bin": b""} + unit = {"file": "empty.bin", "offset": 0, "length": 0, + "sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"} + self.assertTrue(self.verify(dict(ver, units=[unit]), files, plan, op)) + self.assertFalse(self.verify(dict(ver, units=[]), files, plan, op)) + self.assertFalse(self.verify(dict(ver, units=[unit, unit]), files, plan, op)) + self.assertFalse(self.verify(dict(ver, units=[dict(unit, offset=1)]), files, plan, op)) + unit_plan = dict(plan, mode="units", units=[unit]) + self.assertFalse(self.verify(dict(ver, mode="units", units=[unit]), files, unit_plan, op)) + + def test_reload_apply_verifies_empty_file(self): + self.verify_reload_empty_file("apply") + + def test_reload_rollback_verifies_empty_file(self): + self.verify_reload_empty_file("rollback") + def test_units_rollback_requires_all_planned_base_intervals(self): _, plan, ver = self.fixtures(op="rollback") base = {"model.bin": b"012345", "config.json": b"old"}