build(agent): jabba#56a767 iteration
This commit is contained in:
parent
82b5731339
commit
d68bdeab56
|
|
@ -0,0 +1,106 @@
|
|||
"""Small CLI for Mission CSFD operations.
|
||||
|
||||
Usage:
|
||||
python -m missionledger.cli generate --plandelta plan.json
|
||||
python -m missionledger.cli inspect --plandelta plan.json
|
||||
python -m missionledger.cli sign --plandelta plan.json --keyfile key.bin
|
||||
python -m missionledger.cli verify --plandelta plan.json --signature SIGHEX --vkfile vk.bin
|
||||
|
||||
This is a lightweight tool meant for demos and tests. For production use,
|
||||
integrate with a proper key management system and secure proof stores.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
from .dsl import PlanDelta
|
||||
from .csfd import generate_csfd, sign_csfd, verify_csfd_signature
|
||||
|
||||
|
||||
def load_plandelta(path: str) -> PlanDelta:
|
||||
with open(path, "rb") as f:
|
||||
data = json.load(f)
|
||||
# allow plain dicts matching PlanDelta fields
|
||||
return PlanDelta(delta=data.get("delta", {}), version=int(data.get("version", 0)), metadata=data.get("metadata", {}))
|
||||
|
||||
|
||||
def cmd_generate(args: argparse.Namespace) -> int:
|
||||
pd = load_plandelta(args.plandelta)
|
||||
csfd = generate_csfd(pd, signer=args.signer or "", n_frames=args.nframes)
|
||||
print(json.dumps({"frames": [f.__dict__ for f in csfd.frames], "signer": csfd.signer, "digest_hex": csfd.digest_hex}, sort_keys=True, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_inspect(args: argparse.Namespace) -> int:
|
||||
# inspect is same as generate but pretty-prints
|
||||
pd = load_plandelta(args.plandelta)
|
||||
csfd = generate_csfd(pd, signer=args.signer or "", n_frames=args.nframes)
|
||||
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 cmd_sign(args: argparse.Namespace) -> int:
|
||||
pd = load_plandelta(args.plandelta)
|
||||
csfd = generate_csfd(pd, signer=args.signer or "", n_frames=args.nframes)
|
||||
# read key bytes
|
||||
with open(args.keyfile, "rb") as f:
|
||||
keyb = f.read()
|
||||
sig = sign_csfd(csfd, keyb)
|
||||
out = {"signature": sig, "digest_hex": csfd.digest_hex}
|
||||
print(json.dumps(out, sort_keys=True, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_verify(args: argparse.Namespace) -> int:
|
||||
pd = load_plandelta(args.plandelta)
|
||||
csfd = generate_csfd(pd, signer=args.signer or "", n_frames=args.nframes)
|
||||
with open(args.vkfile, "rb") as f:
|
||||
vk = f.read()
|
||||
ok = verify_csfd_signature(csfd, args.signature, vk)
|
||||
print(json.dumps({"ok": bool(ok)}))
|
||||
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")
|
||||
|
||||
gen = sub.add_parser("generate")
|
||||
gen.add_argument("--plandelta", required=True)
|
||||
gen.add_argument("--signer", required=False)
|
||||
gen.add_argument("--nframes", type=int, default=3)
|
||||
gen.set_defaults(func=cmd_generate)
|
||||
|
||||
ins = sub.add_parser("inspect")
|
||||
ins.add_argument("--plandelta", required=True)
|
||||
ins.add_argument("--signer", required=False)
|
||||
ins.add_argument("--nframes", type=int, default=3)
|
||||
ins.set_defaults(func=cmd_inspect)
|
||||
|
||||
signp = sub.add_parser("sign")
|
||||
signp.add_argument("--plandelta", required=True)
|
||||
signp.add_argument("--keyfile", required=True)
|
||||
signp.add_argument("--signer", required=False)
|
||||
signp.add_argument("--nframes", type=int, default=1)
|
||||
signp.set_defaults(func=cmd_sign)
|
||||
|
||||
ver = sub.add_parser("verify")
|
||||
ver.add_argument("--plandelta", required=True)
|
||||
ver.add_argument("--signature", required=True)
|
||||
ver.add_argument("--vkfile", required=True)
|
||||
ver.add_argument("--signer", required=False)
|
||||
ver.add_argument("--nframes", type=int, default=1)
|
||||
ver.set_defaults(func=cmd_verify)
|
||||
|
||||
args = p.parse_args(argv)
|
||||
if not hasattr(args, "func"):
|
||||
p.print_help()
|
||||
return 2
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -0,0 +1,65 @@
|
|||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def write_temp(data: dict):
|
||||
fd, path = tempfile.mkstemp(text=True)
|
||||
with os.fdopen(fd, "w") as f:
|
||||
json.dump(data, f)
|
||||
return path
|
||||
|
||||
|
||||
def test_cli_generate_and_inspect():
|
||||
from missionledger.cli import main
|
||||
|
||||
pd = {"delta": {"move": True}, "version": 1, "metadata": {"invariants": {"a": True}}}
|
||||
path = write_temp(pd)
|
||||
|
||||
# generate prints a compact JSON
|
||||
rc = main(["generate", "--plandelta", path, "--nframes", "2", "--signer", "cli-test"])
|
||||
assert rc == 0
|
||||
|
||||
rc = main(["inspect", "--plandelta", path, "--nframes", "2", "--signer", "cli-test"])
|
||||
assert rc == 0
|
||||
|
||||
|
||||
def test_cli_sign_and_verify_roundtrip():
|
||||
pytest.importorskip("nacl")
|
||||
|
||||
from nacl.signing import SigningKey
|
||||
from missionledger.cli import main
|
||||
|
||||
pd = {"delta": {"x": 1}, "version": 1, "metadata": {"invariants": {"a": True}}}
|
||||
path = write_temp(pd)
|
||||
|
||||
# create keypair files
|
||||
sk = SigningKey.generate()
|
||||
vk = sk.verify_key
|
||||
|
||||
keyfile = tempfile.NamedTemporaryFile(delete=False)
|
||||
keyfile.write(sk.encode())
|
||||
keyfile.close()
|
||||
|
||||
vkfile = tempfile.NamedTemporaryFile(delete=False)
|
||||
vkfile.write(vk.encode())
|
||||
vkfile.close()
|
||||
|
||||
# call sign via CLI (prints JSON)
|
||||
rc = main(["sign", "--plandelta", path, "--keyfile", keyfile.name, "--nframes", "1"])
|
||||
assert rc == 0
|
||||
|
||||
# simulate capturing signature by re-generating and signing directly
|
||||
# Use the CLI verify command: first we need the signature string
|
||||
# We'll sign here to get the signature for verification step
|
||||
from missionledger.csfd import generate_csfd, sign_csfd
|
||||
from missionledger.dsl import PlanDelta
|
||||
|
||||
pd_obj = PlanDelta(delta=pd["delta"], version=pd["version"], metadata=pd["metadata"])
|
||||
csfd = generate_csfd(pd_obj, signer="cli-test", n_frames=1)
|
||||
sig = sign_csfd(csfd, sk.encode())
|
||||
|
||||
rc = main(["verify", "--plandelta", path, "--signature", sig, "--vkfile", vkfile.name])
|
||||
assert rc == 0
|
||||
Loading…
Reference in New Issue