build(agent): jabba#56a767 iteration
This commit is contained in:
parent
d68bdeab56
commit
ff98380a9a
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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", ""))
|
||||
|
|
|
|||
|
|
@ -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 = ["."]
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue