From ff98380a9ab10a91d8e97c58e60c6932286dc708 Mon Sep 17 00:00:00 2001 From: agent-56a7678c6cd71659 Date: Thu, 30 Apr 2026 09:07:58 +0200 Subject: [PATCH] build(agent): jabba#56a767 iteration --- missionledger/cli.py | 32 ++++++++++++++++++++++++++++++++ missionledger/csfd.py | 33 +++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_csfd.py | 18 ++++++++++++++++++ 4 files changed, 84 insertions(+), 1 deletion(-) diff --git a/missionledger/cli.py b/missionledger/cli.py index 14210ad..eb8256b 100644 --- a/missionledger/cli.py +++ b/missionledger/cli.py @@ -18,6 +18,7 @@ from typing import Optional from .dsl import PlanDelta from .csfd import generate_csfd, sign_csfd, verify_csfd_signature +from .csfd import csfd_to_cbor, csfd_from_cbor def load_plandelta(path: str) -> PlanDelta: @@ -64,6 +65,26 @@ def cmd_verify(args: argparse.Namespace) -> int: return 0 +def cmd_export_cbor(args: argparse.Namespace) -> int: + pd = load_plandelta(args.plandelta) + csfd = generate_csfd(pd, signer=args.signer or "", n_frames=args.nframes) + b = csfd_to_cbor(csfd) + # write bytes to file + with open(args.outfile, "wb") as f: + f.write(b) + print(json.dumps({"outfile": args.outfile, "digest_hex": csfd.digest_hex})) + return 0 + + +def cmd_import_cbor(args: argparse.Namespace) -> int: + # read bytes and print inspected CSFD + with open(args.infile, "rb") as f: + b = f.read() + csfd = csfd_from_cbor(b) + print(json.dumps({"frames": [f.__dict__ for f in csfd.frames], "signer": csfd.signer, "digest_hex": csfd.digest_hex}, indent=2, sort_keys=True)) + return 0 + + def main(argv: Optional[list[str]] = None) -> int: p = argparse.ArgumentParser(prog="mission-csfd", description="Mission CSFD helper CLI (demo)") sub = p.add_subparsers(dest="cmd") @@ -95,6 +116,17 @@ def main(argv: Optional[list[str]] = None) -> int: ver.add_argument("--nframes", type=int, default=1) ver.set_defaults(func=cmd_verify) + exp = sub.add_parser("export-cbor") + exp.add_argument("--plandelta", required=True) + exp.add_argument("--outfile", required=True) + exp.add_argument("--signer", required=False) + exp.add_argument("--nframes", type=int, default=3) + exp.set_defaults(func=cmd_export_cbor) + + imp = sub.add_parser("import-cbor") + imp.add_argument("--infile", required=True) + imp.set_defaults(func=cmd_import_cbor) + args = p.parse_args(argv) if not hasattr(args, "func"): p.print_help() diff --git a/missionledger/csfd.py b/missionledger/csfd.py index 8c15980..36b5084 100644 --- a/missionledger/csfd.py +++ b/missionledger/csfd.py @@ -17,6 +17,13 @@ except Exception: BadSignatureError = Exception _HAS_PYNACL = False +try: + import cbor2 + _HAS_CBOR = True +except Exception: + cbor2 = None + _HAS_CBOR = False + from .dsl import PlanDelta @@ -142,3 +149,29 @@ def verify_csfd_signature(csfd: CSFDDigest, signature_hex: str, verify_key_bytes return True except BadSignatureError: return False + + +def csfd_to_cbor(csfd: CSFDDigest) -> bytes: + """Serialize a CSFD to CBOR bytes. Requires cbor2 if you want CBOR output. + + Falls back to JSON bytes if cbor2 is not available. + """ + obj = asdict(csfd) + if _HAS_CBOR: + return cbor2.dumps(obj) + # fallback: compact JSON bytes + return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode() + + +def csfd_from_cbor(b: bytes) -> CSFDDigest: + """Deserialize CBOR/JSON bytes into a CSFDDigest. + + Accepts CBOR if cbor2 is available; otherwise expects UTF-8 JSON bytes. + """ + if _HAS_CBOR: + obj = cbor2.loads(b) + else: + obj = json.loads(b.decode()) + + frames = [CSFDFrame(**f) for f in obj.get("frames", [])] + return CSFDDigest(frames=frames, signer=obj.get("signer", ""), digest_hex=obj.get("digest_hex", "")) diff --git a/pyproject.toml b/pyproject.toml index 72852d5..49ead4e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ description = "MVP core for verifiable contract-driven mission planning in inter readme = "README.md" requires-python = ">=3.8" license = {text = "MIT"} -dependencies = ["pynacl>=1.4.0"] +dependencies = ["pynacl>=1.4.0", "cbor2>=5.0.0"] [tool.setuptools.packages.find] where = ["."] diff --git a/tests/test_csfd.py b/tests/test_csfd.py index 35a03ae..bd7c6ec 100644 --- a/tests/test_csfd.py +++ b/tests/test_csfd.py @@ -56,3 +56,21 @@ def test_csfd_signing_and_verification(): vk = sk.verify_key ok = verify_csfd_signature(csfd, sig_hex, vk.encode()) assert ok is True + + +def test_csfd_cbor_roundtrip(): + import pytest + pytest.importorskip("cbor2") + + from missionledger.csfd import csfd_to_cbor, csfd_from_cbor + + pd = PlanDelta(delta={"y": 2}, version=1, metadata={"invariants": {"a": True}}) + csfd = generate_csfd(pd, signer="s", n_frames=2) + + b = csfd_to_cbor(csfd) + assert isinstance(b, (bytes, bytearray)) + + csfd2 = csfd_from_cbor(b) + assert csfd2.digest_hex == csfd.digest_hex + assert csfd2.signer == csfd.signer + assert len(csfd2.frames) == len(csfd.frames)