coulomb/ops-mason
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare typed structural evidence without credential success circularity

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 02:42:57 +02:00
parent 247c829569
commit f52c94cd4e
7 changed files with 1094 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

138
src/ops_mason/cli.py
View file

@ -37,24 +37,51 @@ def _parser() -> argparse.ArgumentParser:
"--expect-digest", required=True, help="exact digest returned by preflight"
)
apply_parser.add_argument("--readiness-repo", type=Path)
apply_parser.add_argument("--activation", choices=("APPROVED", "BREAK_GLASS"), default="APPROVED")
apply_parser.add_argument(
"--activation", choices=("APPROVED", "BREAK_GLASS"), default="APPROVED"
)
apply_parser.add_argument("--break-glass-reason")
plan = families.add_parser("plan", help="construction plan derived from a SecurityCase")
plan = families.add_parser(
"plan", help="construction plan derived from a SecurityCase"
)
plan.add_argument("--case", required=True, type=Path)
plan.add_argument("--out", type=Path, help="write the plan here (default: stdout)")
plan.add_argument("--no-live", action="store_true",
help="skip the live OpenBao survey (marks the plan's survey source 'none')")
plan.add_argument(
"--no-live",
action="store_true",
help="skip the live OpenBao survey (marks the plan's survey source 'none')",
)
plan.add_argument("--netkingdom-root", type=Path, default=None)
plan.add_argument("--allow-unimplemented", action="store_true",
help="reference case only: render against an acknowledged type")
build = families.add_parser("build", help="build bundle from an accepted SecurityCase and its reviewed plan")
plan.add_argument(
"--allow-unimplemented",
action="store_true",
help="reference case only: render against an acknowledged type",
)
build = families.add_parser(
"build", help="build bundle from an accepted SecurityCase and its reviewed plan"
)
build.add_argument("--case", required=True, type=Path)
build.add_argument("--plan", required=True, type=Path, help="the reviewed plan from plan --case")
build.add_argument("--out", type=Path, help="bundle directory (default: builds/case-<id>)")
build.add_argument("--allow-unimplemented", action="store_true",
help="reference case only: build against an acknowledged type")
build.add_argument(
"--plan", required=True, type=Path, help="the reviewed plan from plan --case"
)
build.add_argument(
"--out", type=Path, help="bundle directory (default: builds/case-<id>)"
)
build.add_argument(
"--allow-unimplemented",
action="store_true",
help="reference case only: build against an acknowledged type",
)
build.add_argument("--netkingdom-root", type=Path, default=None)
structure = families.add_parser(
"structure-request", help="read-only typed structural evidence request"
)
structure.add_argument("--case", required=True, type=Path)
structure.add_argument("--plan", required=True, type=Path)
structure.add_argument("--bundle", required=True, type=Path)
structure.add_argument("--netkingdom-root", type=Path)
structure.add_argument("--allow-unimplemented", action="store_true")
return parser
@ -63,15 +90,39 @@ def _build(args) -> int:
nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT
try:
case_id = (case_plan.yaml.safe_load(args.case.read_text()) or {}).get("spec", {}).get("id", "unknown")
case_id = (
(case_plan.yaml.safe_load(args.case.read_text()) or {})
.get("spec", {})
.get("id", "unknown")
)
out = args.out or Path("builds") / f"case-{case_id}"
result = case_build.build_from_case(args.case, args.plan, out, nk_root=nk_root,
allow_unimplemented=args.allow_unimplemented)
except (case_plan.CasePlanError, case_build.CaseBuildRefused, FileExistsError, OSError) as exc:
print(json.dumps({"ok": False, "error": str(exc), "issues": getattr(exc, "issues", [])}), file=sys.stderr)
result = case_build.build_from_case(
args.case,
args.plan,
out,
nk_root=nk_root,
allow_unimplemented=args.allow_unimplemented,
)
except (
case_plan.CasePlanError,
case_build.CaseBuildRefused,
FileExistsError,
OSError,
) as exc:
print(
json.dumps(
{"ok": False, "error": str(exc), "issues": getattr(exc, "issues", [])}
),
file=sys.stderr,
)
return 2
fails = case_build.validate_with_platform([Path(p) for p in result["ccrs"]])
print(json.dumps({"ok": not fails, "platform_validation": fails or "pass", **result}, indent=2))
print(
json.dumps(
{"ok": not fails, "platform_validation": fails or "pass", **result},
indent=2,
)
)
return 1 if fails else 0
@ -81,21 +132,66 @@ def _plan(args) -> int:
nk_root = args.netkingdom_root or case_plan.DEFAULT_NK_ROOT
try:
doc, collisions, rendered = case_plan.plan_from_case(
args.case, nk_root=nk_root, structure=case_plan.Structure() if args.no_live else None,
allow_unimplemented=args.allow_unimplemented)
args.case,
nk_root=nk_root,
structure=case_plan.Structure() if args.no_live else None,
allow_unimplemented=args.allow_unimplemented,
)
except case_plan.CasePlanError as exc:
print(json.dumps({"ok": False, "error": str(exc), "issues": exc.issues}), file=sys.stderr)
print(
json.dumps({"ok": False, "error": str(exc), "issues": exc.issues}),
file=sys.stderr,
)
return 2
if args.out:
args.out.write_text(doc)
else:
sys.stdout.write(doc)
print(json.dumps({"ok": not collisions, "digest": rendered["digest"], "collisions": collisions}), file=sys.stderr)
print(
json.dumps(
{
"ok": not collisions,
"digest": rendered["digest"],
"collisions": collisions,
}
),
file=sys.stderr,
)
return 1 if collisions else 0
def main(argv: Sequence[str] | None = None) -> int:
args = _parser().parse_args(argv)
if args.family == "structure-request":
from ops_mason import case_structure, case_plan
try:
request = case_structure.prepare(
args.case,
args.plan,
args.bundle,
nk_root=args.netkingdom_root or case_plan.DEFAULT_NK_ROOT,
allow_unimplemented=args.allow_unimplemented,
)
except Exception:
print(
json.dumps(
{"ok": False, "error": "structural-source-preparation-refused"}
),
file=sys.stderr,
)
return 2
print(
json.dumps(
{
"source_only": True,
"request_digest": request.fingerprint,
"request": request.as_dict(),
},
indent=2,
)
)
return 0
if args.family == "plan":
return _plan(args)
if args.family == "build":