From f52c94cd4e36276d58cd23c34be6d22f02682ba4 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 11 Oct 2026 02:42:57 +0200 Subject: [PATCH] Prepare typed structural evidence without credential success circularity Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade --- ...jo-structural-materialization-source-v1.md | 117 ++++ src/ops_mason/case_structure.py | 512 ++++++++++++++++++ src/ops_mason/cli.py | 138 ++++- tests/test_case_build.py | 4 +- tests/test_case_plan.py | 3 +- tests/test_case_structure.py | 324 +++++++++++ ...N-WP-0010-security-case-materialization.md | 23 +- 7 files changed, 1094 insertions(+), 27 deletions(-) create mode 100644 docs/forgejo-structural-materialization-source-v1.md create mode 100644 src/ops_mason/case_structure.py create mode 100644 tests/test_case_structure.py diff --git a/docs/forgejo-structural-materialization-source-v1.md b/docs/forgejo-structural-materialization-source-v1.md new file mode 100644 index 0000000..9175c5c --- /dev/null +++ b/docs/forgejo-structural-materialization-source-v1.md @@ -0,0 +1,117 @@ +# Forgejo structural materialization preparation v1 — 2026-10-11 + +Materialization is the policies, roles, fixed custody layout and catalog identity +under accepted SecurityCaseTypes §2/§7. It precedes follow-through. Minting a +secret, delivering it, registering a client and verifying the native consumer +are follow-through success, not structural prerequisites. The accepted execution +architecture (memo infd-20261010-security-case-execution version1, confirmed +approval e46b149b-ec3c-4b3b-98a7-dcb4934448ca) preserves this split and requires +independent structural evidence. It approves no new native session or type. + +## Concrete source drift and the hash boundary + +The committed Forgejo case and bundle are both @1.1; all bundle file hashes still +match. Both Mason plan and manifest pin full render af1f4f81ef313f5e086e6e3de17197d9112dd40f00b4c20b52b5ca73c7a6d18a. +Re-rendering type e2174ce with today's exact validator/provider declarations +reproduces that digest. Current type patch 35880bb adds --check-config and +startup-check wording to register-client; it renders full digest +e64b56e4590b57b858303fcda1a474693e3dd821bc83df26646fe78fc345498a. +Only follow_through changes. Materialization, composition, verification and +rollback compare equal. This is a real stricter execution guard, not volatile +Git/time data. The complete run digest must remain different. + +`src/ops_mason/case_structure.py` prepares a separate structural projection +fingerprint and exact current case/type/plan/bundle pins. Before producing a +request it verifies every hashed bundle file, the complete current generated +policy (including self lifecycle), exact CCR reader/role/target bindings, custody +layout inventory and catalog source. A freshly rehashed wider ACL, namespace or +sibling-path change still fails against current templates. Non-semantic HCL +header comments may differ; policy statements must match. Unknown bundle files, +path traversal and symlinks are refused. The stale full run digest is retained as +reviewed_render_digest with run_source_changed=true, never promoted into current +execution acceptance. Source structural equality is distinct from native presence +or creation ownership. + +Read-only command, using NetKingdom's validator/render and no live survey: + +``` +ops-mason structure-request --case ../net-kingdom/security-cases/forgejo-netkingdom-oidc.yaml --plan plans/case-forgejo-netkingdom-oidc.md --bundle builds/case-forgejo-netkingdom-oidc --allow-unimplemented +``` + +The real source yields five object requests: fixed KV layout, ESO policy, +consumer policy, ESO role and draft catalog pointer. Native policy/role evidence +belongs to RailiancePlatform; the pointer source belongs to OpsWarden. KV layout +proof is metadata/layout only and never requires or claims a value. Neither mint +nor secret-delivered is in this structural inventory. Each source input is bounded +at 256 KiB; maximum inventory is 32 objects/64 files. + +## Typed receipt seam, not a bare status label + +Request protocol ops-mason.case-structure-request.v1 and typed Observation are a +local source seam; owner acknowledgment is still required before wiring them into +native dispatch. Each observation binds the full request, object ID, owner, +exact target and expected structural fingerprint, disposition and evidence ref. +The injected verifier must authenticate native owner provenance or explicitly +attended manual evidence. It sees the complete observation/change receipt, never +only a trusted JSON ref. No native verifier is installed here. Fake unit verifiers +prove refusal/projection behavior only. + +Native structure is classified created or preexisting-matched. Created requires +an independently verified change receipt. Layout-only and pointer-source neither +claim stored values nor creation undo. VerifiedStructure projects only proven +created typed StructuralObject identities into reverse-order rollback, retaining +object ID, kind, target, owner and source digest. Policy/role name collisions remain +distinct; shared objects and source layout +never acquire delete authority. Provider diagnostics, OS/Python output and raw +exceptions are withheld during receipt verification. This produces candidate +structural evidence, not a case status mutation or global native-complete result. + +`bind_status_transition` previews the raw-hash seam without writing a case: it +re-authenticates all structural observations, requires exact accepted source +bytes, permits only semantic status accepted→materialized with every other field +unchanged, and retains original/request/materialized digests. An admitted owner +must authenticate that binding for Warden; the function supplies no signature, +case update or authority. Parameter, owner or acceptance changes require new +review. This avoids binding an accepted-status receipt to the wrong raw digest +when Warden and the broker later read the materialized case. + +The legacy committed rollback.yaml lists planned identities in removes_exactly; +that list is not creation evidence. Existing case_plan uses it as an own-build +collision hint. The new structural receipt path deliberately does not inherit +that hint as ownership or rollback permission. Replacing the legacy planning hint +with authenticated receipt ingestion is a distinct integration step requiring an +actual owner evidence adapter; source JSON alone cannot discharge it. + +## Exact owner contracts still needed + +| Owner | Existing source capability | Missing machine/native contract | +| --- | --- | --- | +| ops-mason + RailiancePlatform | Source bundle, CCR41 ESO structure applied; typed metadata inventory prepared | Fresh independently authenticated policy/role/layout observations; created vs preexisting provenance; exact receipt signing/attended manual custody and binding consumed by Warden | +| railiance-apps + RailiancePlatform | CCR42 consumer policy draft | Consumer reader openbao.auth, actual namespace/SA or admitted identity, review and application; CCR42 is still in_flight with historical @1.0 provenance and must be reconciled explicitly to current @1.1 before using its receipt | +| RailiancePlatform | Exact generated-in-custody broker, CAS0, materialized-case/client checks, grant_token_revoked receipt | Closed adapter interface/source acceptance; immutable case source resolution, per-attempt owner receipt plus broker-grant and parent-session revocation fingerprints; owner-native CAS already-present reconciliation (never rerun generation) | +| NetKingdom + KeyCape | Contained registration helper with --check-config, expected UID/RV, byte-preserving merge/readback | Typed adapter parameters and receipt/undo contract for groups/env/client registration, source-resolved current case, native UID/RV refusal and attribution; no user-engine runtime-token substitution | +| RailiancePlatform | Delivery source for ESO | Fixed metadata-only delivery/status/undo adapter and independently verified created target refs; readiness separate from materialization | +| railiance-apps | Forgejo OIDC/group-enforcement consumer ownership | Fixed owner adapter or authenticated receipt-only step, exact CCR42 binding, actual consumer group refusal and rollback evidence | +| NetKingdom + OpsWarden | Acknowledged descriptive type1.1; durable local execution kernel | Accepted exact machine descriptors and owner adapter/source refs pinned in the type, all ordered verification steps, native evidence/caller provenance, contained session integration and dispatch CLI | + +The first broker descriptor must require both broker-grant and parent-session +revocation scopes, exact case/type/source/catalog/actor/lane pins and per-attempt +identity. Existing broker exit3 already-present is owner reconciliation work, +not verified failed-before-change/retry-safe. Current descriptive type remains +plan-only. No prose entry_point/check/action is interpreted as shell or argv. + +## Validation and remaining state + +Thirty-three relevant tests pass: fifteen structural/CLI tests plus existing case +plan/build suites. Structural negatives alter and rehash source ACLs, auth +namespace, catalog and KV inventory; receipt negatives omit owners, forge exact +pins or use secret-minted as structural status. Created/preexisting projection +excludes shared objects from undo; source CLI performs no live survey or bundle +write. Existing stale @1.0 test assertions now validate the accepted input type +instead of falsely failing the @1.1 reference case. Ruff checks pass. + +MASON-WP-0010-T03 remains progress for actual structural evidence and CCR42 +binding/application; mint is removed from its structural readiness blocker. +WARDEN-WP-0044-T03 and NK-WP-0048-T06 retain native follow-through, full current +run digest/machine source binding, verification and first live-case acceptance. +No case/type status, credential, policy, catalog activation or workload was changed. diff --git a/src/ops_mason/case_structure.py b/src/ops_mason/case_structure.py new file mode 100644 index 0000000..61a6e05 --- /dev/null +++ b/src/ops_mason/case_structure.py @@ -0,0 +1,512 @@ +"""Source-only structural evidence preparation; no native materialization act. + +A construction bundle is not an ownership or native presence receipt. This +module pins it and produces the bounded metadata contract an owner must attest. +Credential mint/delivery and activation never become structural prerequisites. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from contextlib import contextmanager, redirect_stdout, redirect_stderr +import hashlib +import json +import os +import threading +from pathlib import Path, PurePosixPath +import re +from typing import Callable + +import yaml +from ops_mason import case_build, case_plan + +LIMIT = 262144 +OUTPUT_LOCK = threading.RLock() +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,191}\Z") +DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z") + + +class StructureRefused(ValueError): + """Metadata is incomplete, unverified or not bound to the reviewed structure.""" + + +class _Discard: + def write(self, text): + return len(text) + + def flush(self): + pass + + +@contextmanager +def quiet_owner(): + """Authenticate metadata without retaining owner diagnostics or inherited I/O.""" + with OUTPUT_LOCK: + saved = [] + null = os.open(os.devnull, os.O_WRONLY) + try: + for fd in (1, 2): + saved.append((fd, os.dup(fd))) + os.dup2(null, fd) + with redirect_stdout(_Discard()), redirect_stderr(_Discard()): + try: + yield + except BaseException: + raise StructureRefused( + "owner structural evidence verification failed" + ) from None + finally: + for fd, original in saved: + os.dup2(original, fd) + os.close(original) + os.close(null) + + +def sha(source: bytes) -> str: + return "sha256:" + hashlib.sha256(source).hexdigest() + + +def read(path: Path) -> bytes: + if path.is_symlink() or path.stat().st_size > LIMIT: + raise StructureRefused("source path is unsafe or exceeds limit") + return path.read_bytes() + + +@dataclass(frozen=True) +class StructuralObject: + id: str + kind: str + target: str + expected_digest: str + evidence_owner: str + proof: str + + +@dataclass(frozen=True) +class Request: + schema: str + case: str + case_digest: str + type_ref: str + type_digest: str + render_digest: str + plan_digest: str + bundle_digest: str + structural_digest: str + reviewed_render_digest: str + run_source_changed: bool + objects: tuple[StructuralObject, ...] + + def as_dict(self): + return asdict(self) + + @property + def fingerprint(self): + return sha( + json.dumps(self.as_dict(), sort_keys=True, separators=(",", ":")).encode() + ) + + +def prepare( + case_path: Path, + plan_path: Path, + bundle_path: Path, + *, + nk_root=case_plan.DEFAULT_NK_ROOT, + types_root=None, + allow_unimplemented=False, +) -> Request: + """Use NetKingdom's validator/render; verify every bundle file before emitting.""" + case_source, plan_source, manifest_source = ( + read(p) for p in (case_path, plan_path, bundle_path / "manifest.json") + ) + case, type_doc, rendered = case_plan.render( + case_path, nk_root, types_root, allow_unimplemented + ) + root = types_root or nk_root / "security-case-types" + type_source = read(root / type_doc["spec"]["id"] / "type.yaml") + if yaml.safe_load(type_source) != type_doc: + raise StructureRefused("type source changed during preparation") + if yaml.safe_load(case_source) != case: + raise StructureRefused("case source changed during preparation") + try: + manifest = json.loads(manifest_source) + front = yaml.safe_load(plan_source.decode().split("---\n", 2)[1]) + provenance = manifest["security_case"] + if ( + front["status"] != "reviewed" + or front["render_digest"] != provenance["render_digest"] + ): + raise ValueError + if (provenance["case"], provenance["type"]) != ( + case["spec"]["id"], + rendered["type"], + ): + raise ValueError + if provenance["acceptance_record"] != case_plan.acceptance_ref(case)["record"]: + raise ValueError + files = manifest["files"] + if not isinstance(files, dict) or not 1 <= len(files) <= 64: + raise ValueError + for relative, expected in files.items(): + name = PurePosixPath(relative) + if ( + name.is_absolute() + or ".." in name.parts + or not name.parts + or name.as_posix() != relative + ): + raise ValueError + target = bundle_path / name + if not target.resolve().is_relative_to(bundle_path.resolve()): + raise ValueError + if sha(read(target)) != "sha256:" + expected: + raise ValueError + except (ValueError, KeyError, TypeError, IndexError, yaml.YAMLError): + raise StructureRefused( + "bundle or reviewed plan does not match fresh source" + ) from None + # Verify the complete current structural footprint against bundle contents, + # independently of full run prose changes. Never weaken the full run digest. + _match_structural_bundle(case, rendered, bundle_path, files) + structural_digest = sha( + json.dumps( + { + "case": case["spec"]["id"], + "type": rendered["type"], + "acceptance": case_plan.acceptance_ref(case), + "materialization": rendered["materialization"], + }, + sort_keys=True, + separators=(",", ":"), + ).encode() + ) + objects = [] + owners = { + "openbao_policy": ("railiance-platform", "native-structure"), + "openbao_role": ("railiance-platform", "native-structure"), + "kv_path": ("railiance-platform", "layout-only"), + "catalog_entry": ("ops-warden", "pointer-source"), + } + for item in rendered["materialization"]: + if item["kind"] not in owners or not all( + IDENTIFIER.fullmatch(item[k]) for k in ("id", "template") + ): + raise StructureRefused("unsupported structural object") + owner, proof = owners[item["kind"]] + expected = sha(json.dumps(item, sort_keys=True, separators=(",", ":")).encode()) + objects.append( + StructuralObject( + item["id"], item["kind"], item["template"], expected, owner, proof + ) + ) + if len(objects) > 32 or len({o.id for o in objects}) != len(objects): + raise StructureRefused( + "structural inventory exceeds bounds or repeats identity" + ) + return Request( + "ops-mason.case-structure-request.v1", + case["spec"]["id"], + sha(case_source), + rendered["type"], + sha(type_source), + rendered["digest"], + sha(plan_source), + sha(manifest_source), + structural_digest, + provenance["render_digest"], + provenance["render_digest"] != rendered["digest"], + tuple(objects), + ) + + +def _match_structural_bundle(case, rendered, bundle_path, files): + """Exact generated structure, including self lifecycle; source is not ownership.""" + + def hcl(text): + return "\n".join( + line.strip() + for line in text.splitlines() + if line.strip() and not line.lstrip().startswith("#") + ) + + kv = [item for item in rendered["materialization"] if item["kind"] == "kv_path"] + if len(kv) != 1: + raise StructureRefused( + "builder structural contract requires one custody layout" + ) + prov = case_build._provenance(case, rendered) + roles = { + yaml.safe_load(item["body"]).get("token_policies", [None])[0]: item + for item in rendered["materialization"] + if item["kind"] == "openbao_role" + } + required = {"rollback.yaml"} + for item in rendered["materialization"]: + if item["kind"] == "openbao_policy": + relative = f"openbao/policies/{item['template']}.hcl" + required.add(relative) + if hcl(read(bundle_path / relative).decode()) != hcl( + case_build._policy_hcl(item, prov) + ): + raise StructureRefused( + "current structural policy differs from prepared bundle" + ) + expected = case_build._ccr( + case, + rendered, + item, + roles.get(item["template"]), + kv[0]["template"], + prov, + ) + relative = f"credential-change-requests/{expected['id']}.yaml" + required.add(relative) + actual = yaml.safe_load(read(bundle_path / relative)) + if ( + actual.get("openbao") != expected["openbao"] + or actual.get("target") != expected["target"] + or actual.get("request_type") != expected["request_type"] + ): + raise StructureRefused( + "current role or reader binding differs from prepared bundle" + ) + elif item["kind"] == "catalog_entry": + required.add("catalog-proposal.yaml") + if ( + yaml.safe_load(read(bundle_path / "catalog-proposal.yaml")).get("entry") + != item["body"] + ): + raise StructureRefused( + "current catalog pointer differs from prepared bundle" + ) + if set(files) != required: + raise StructureRefused( + "bundle inventory contains missing or unsupported source files" + ) + rollback = yaml.safe_load(read(bundle_path / "rollback.yaml")) + expected = [ + {"kind": item["kind"], "name": item["template"]} + for item in reversed(rendered["materialization"]) + ] + if rollback.get("removes_exactly") != expected: + raise StructureRefused( + "bundle structural inventory differs from current templates" + ) + # These are planned identities, deliberately not creation/undo authority. + + +@dataclass(frozen=True, slots=True) +class Observation: + object_id: str + owner: str + target: str + expected_digest: str + disposition: str + evidence: str + change_receipt: str | None = None + + +@dataclass(frozen=True) +class VerifiedStructure: + request_digest: str + created: tuple[StructuralObject, ...] + preexisting: tuple[StructuralObject, ...] + layout_only: tuple[StructuralObject, ...] + pointers: tuple[StructuralObject, ...] + evidence_fingerprints: tuple[str, ...] + + @property + def rollback_targets(self): + return tuple(reversed(self.created)) + + +def verify( + request: Request, + observations: tuple[Observation, ...], + verifier: Callable[[Request, StructuralObject, Observation], bool], +) -> VerifiedStructure: + """Independent authenticated owner/manual evidence required, not trusted JSON. + + Returns candidate structural evidence only. No case status or native execution + receipt is issued; this local contract still needs owner seam acknowledgment. + """ + if ( + type(request) is not Request + or request.schema != "ops-mason.case-structure-request.v1" + ): + raise StructureRefused("unknown structural request protocol") + if not all( + isinstance(value, str) and DIGEST.fullmatch(value) + for value in ( + request.case_digest, + request.type_digest, + request.render_digest, + request.plan_digest, + request.bundle_digest, + request.structural_digest, + request.reviewed_render_digest, + ) + ): + raise StructureRefused("request source pins are malformed") + if ( + type(request.case) is not str + or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,127}", request.case) + or type(request.type_ref) is not str + or not re.fullmatch( + r"[a-z0-9][a-z0-9-]{0,127}@[0-9]+\.[0-9]+", request.type_ref + ) + ): + raise StructureRefused("request case or type identifier malformed") + if type(request.run_source_changed) is not bool or request.run_source_changed != ( + request.render_digest != request.reviewed_render_digest + ): + raise StructureRefused("execution source change flag disagrees with exact pins") + kinds = { + "openbao_policy": ("railiance-platform", "native-structure"), + "openbao_role": ("railiance-platform", "native-structure"), + "kv_path": ("railiance-platform", "layout-only"), + "catalog_entry": ("ops-warden", "pointer-source"), + } + if type(request.objects) is not tuple or not 1 <= len(request.objects) <= 32: + raise StructureRefused("unknown structural request inventory") + for obj in request.objects: + if type(obj) is not StructuralObject or kinds.get(obj.kind) != ( + obj.evidence_owner, + obj.proof, + ): + raise StructureRefused("unknown structural proof kind or owner") + if not all( + isinstance(value, str) and IDENTIFIER.fullmatch(value) + for value in (obj.id, obj.target) + ) or not DIGEST.fullmatch(obj.expected_digest): + raise StructureRefused("structural metadata or source pin malformed") + if len({obj.id for obj in request.objects}) != len(request.objects): + raise StructureRefused("duplicate structural object identity") + if type(observations) is not tuple or len(observations) != len(request.objects): + raise StructureRefused( + "every structural object needs an exact owner observation" + ) + by_id = {o.object_id: o for o in observations if type(o) is Observation} + if len(by_id) != len(observations): + raise StructureRefused("unknown or duplicate observation") + classified = { + "created": [], + "preexisting-matched": [], + "layout-only": [], + "pointer-source": [], + } + fingerprints = [] + for obj in request.objects: + observation = by_id.get(obj.id) + if observation is None or ( + observation.owner, + observation.target, + observation.expected_digest, + ) != (obj.evidence_owner, obj.target, obj.expected_digest): + raise StructureRefused("owner observation scope or source pin mismatch") + if not isinstance(observation.evidence, str) or not IDENTIFIER.fullmatch( + observation.evidence + ): + raise StructureRefused("evidence must be a bounded metadata reference") + permitted = ( + ("created", "preexisting-matched") + if obj.proof == "native-structure" + else (obj.proof,) + ) + if observation.disposition not in permitted: + raise StructureRefused( + "credential success or status labels are not structural proof" + ) + if observation.disposition == "created": + if not isinstance( + observation.change_receipt, str + ) or not IDENTIFIER.fullmatch(observation.change_receipt): + raise StructureRefused( + "creation requires independently verified change receipt" + ) + elif observation.change_receipt is not None: + raise StructureRefused("non-created object cannot carry undo authority") + # The provider must authenticate the complete request/object/observation, + # including manual attender identity or owner transport, not just its ref. + try: + with quiet_owner(): + authenticated = verifier(request, obj, observation) is True + except Exception: + raise StructureRefused( + "owner structural evidence verification failed" + ) from None + if not authenticated: + raise StructureRefused("owner structural evidence unverified") + classified[observation.disposition].append(obj) + fingerprints.append( + sha(json.dumps(asdict(observation), sort_keys=True).encode()) + ) + return VerifiedStructure( + request.fingerprint, + tuple(classified["created"]), + tuple(classified["preexisting-matched"]), + tuple(classified["layout-only"]), + tuple(classified["pointer-source"]), + tuple(fingerprints), + ) + + +@dataclass(frozen=True) +class StatusBinding: + request_digest: str + accepted_case_digest: str + materialized_case_digest: str + structural_digest: str + owner_evidence_fingerprints: tuple[str, ...] + created: tuple[StructuralObject, ...] + preexisting: tuple[StructuralObject, ...] + + +def bind_status_transition( + request: Request, + accepted_source: bytes, + materialized_source: bytes, + observations: tuple[Observation, ...], + verifier: Callable[[Request, StructuralObject, Observation], bool], +) -> StatusBinding: + """Preview a status-only raw-hash binding after independent structural proof. + + No file is written and no proof is signed. An admitted owner must authenticate + this binding before Warden accepts it for its new raw case digest. + """ + if ( + type(accepted_source) is not bytes + or type(materialized_source) is not bytes + or max(len(accepted_source), len(materialized_source)) > LIMIT + ): + raise StructureRefused("case source exceeds bounded transition contract") + if sha(accepted_source) != request.case_digest: + raise StructureRefused("accepted case source pin changed") + try: + before, after = ( + yaml.safe_load(accepted_source), + yaml.safe_load(materialized_source), + ) + if ( + before["spec"]["status"] != "accepted" + or after["spec"]["status"] != "materialized" + ): + raise ValueError + before["spec"]["status"] = "materialized" + if before != after: + raise ValueError + except (yaml.YAMLError, KeyError, TypeError, ValueError): + raise StructureRefused( + "only accepted-to-materialized status change is permitted" + ) from None + verified = verify(request, observations, verifier) + return StatusBinding( + request.fingerprint, + request.case_digest, + sha(materialized_source), + request.structural_digest, + verified.evidence_fingerprints, + verified.created, + verified.preexisting, + ) diff --git a/src/ops_mason/cli.py b/src/ops_mason/cli.py index 9370879..2f0210a 100644 --- a/src/ops_mason/cli.py +++ b/src/ops_mason/cli.py @@ -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-)") - 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-)" + ) + 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": diff --git a/tests/test_case_build.py b/tests/test_case_build.py index 2882be2..f182030 100644 --- a/tests/test_case_build.py +++ b/tests/test_case_build.py @@ -44,7 +44,7 @@ def test_builds_bundle_naming_case_type_and_acceptance(tmp_path, types_root): case = _case(tmp_path) out, log = tmp_path / "bundle", tmp_path / "audit.jsonl" r = cb.build_from_case(case, _plan(tmp_path, case, types_root), out, types_root=types_root, audit_log=log) - assert r["case"] == "forgejo-netkingdom-oidc" and r["type"] == "oidc-browser-relying-party@1.0" + assert r["case"] == "forgejo-netkingdom-oidc" and r["type"] == yaml.safe_load(case.read_text())["spec"]["type"] ccrs = {Path(p).name: yaml.safe_load(Path(p).read_text()) for p in r["ccrs"]} eso = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"] assert eso["status"] == "proposed" and eso["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo" @@ -57,7 +57,7 @@ def test_builds_bundle_naming_case_type_and_acceptance(tmp_path, types_root): assert 'path "platform/data/workloads/keycape/forgejo"' in hcl and '["read"]' in hcl and "forgejo-netkingdom-oidc" in hcl # token_self_lifecycle: true needs the self stanzas, or the platform applier refuses the file. assert 'path "auth/token/lookup-self"' in hcl and 'path "auth/token/revoke-self"' in hcl - stanzas = [l for l in hcl.splitlines() if l.startswith("path ")] + stanzas = [line for line in hcl.splitlines() if line.startswith("path ")] assert len(stanzas) == 3 rollback = yaml.safe_load((out / "rollback.yaml").read_text()) assert rollback["removes_exactly"][-1] == {"kind": "kv_path", "name": "platform/workloads/keycape/forgejo"} diff --git a/tests/test_case_plan.py b/tests/test_case_plan.py index 6a5a0f7..4b083f8 100644 --- a/tests/test_case_plan.py +++ b/tests/test_case_plan.py @@ -1,5 +1,4 @@ import shutil -from pathlib import Path from unittest.mock import MagicMock, patch import pytest @@ -42,7 +41,7 @@ def test_renders_reference_case_as_derived_plan(tmp_path, types_root): doc, collisions, rendered = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=cp.Structure()) assert collisions == [] front = yaml.safe_load(doc.split("---")[1]) - assert front["derived_from_type"] == "oidc-browser-relying-party@1.0" + assert front["derived_from_type"] == yaml.safe_load(CASE.read_text())["spec"]["type"] assert front["render_digest"] == rendered["digest"] and front["status"] == "draft" names = {i["template"] for i in rendered["materialization"]} assert {"platform/workloads/keycape/forgejo", "workload-kv-read-keycape-forgejo-eso", diff --git a/tests/test_case_structure.py b/tests/test_case_structure.py new file mode 100644 index 0000000..6a3967d --- /dev/null +++ b/tests/test_case_structure.py @@ -0,0 +1,324 @@ +"""Source structural projection and fake authenticated owner observations only.""" + +from dataclasses import asdict, replace +import hashlib +import json +import os +from pathlib import Path +import shutil +import pytest +import yaml +from ops_mason import case_plan, case_structure as structure + +ROOT = Path("/home/worsch/ops-mason") +CASE = case_plan.DEFAULT_NK_ROOT / "security-cases/forgejo-netkingdom-oidc.yaml" + + +@pytest.fixture +def prepared(tmp_path): + bundle = tmp_path / "bundle" + shutil.copytree(ROOT / "builds/case-forgejo-netkingdom-oidc", bundle) + plan = tmp_path / "plan.md" + shutil.copy2(ROOT / "plans/case-forgejo-netkingdom-oidc.md", plan) + request = structure.prepare(CASE, plan, bundle, allow_unimplemented=True) + return request, plan, bundle + + +def observations(request): + result = [] + for obj in request.objects: + disposition = "created" if obj.proof == "native-structure" else obj.proof + result.append( + structure.Observation( + obj.id, + obj.evidence_owner, + obj.target, + obj.expected_digest, + disposition, + "fixture-owner-evidence", + "fixture-change-receipt" if disposition == "created" else None, + ) + ) + return tuple(result) + + +def test_actual_first_case_structure_matches_but_full_execution_guard_drift_remains( + prepared, +): + request, _, _ = prepared + assert request.type_ref == "oidc-browser-relying-party@1.1" + assert {o.proof for o in request.objects} == { + "native-structure", + "layout-only", + "pointer-source", + } + assert len(request.objects) == 5 + assert "mint-secret" not in {o.id for o in request.objects} + assert "secret-delivered" not in {o.id for o in request.objects} + + +@pytest.mark.parametrize( + "change", ["extra-policy-rule", "role-binding", "catalog-pointer", "kv-inventory"] +) +def test_structural_mutation_refuses_even_after_attacker_rehashes_manifest( + prepared, change +): + _, plan, bundle = prepared + if change == "extra-policy-rule": + rel = "openbao/policies/workload-kv-read-keycape-forgejo-eso.hcl" + (bundle / rel).write_text( + (bundle / rel).read_text() + + '\npath "platform/data/*" { capabilities = ["read"] }\n' + ) + elif change == "role-binding": + rel = "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml" + data = yaml.safe_load((bundle / rel).read_text()) + data["openbao"]["auth"]["bound_claims"]["service_account_namespaces"] = [ + "other" + ] + (bundle / rel).write_text(yaml.safe_dump(data)) + elif change == "catalog-pointer": + rel = "catalog-proposal.yaml" + data = yaml.safe_load((bundle / rel).read_text()) + data["entry"] += "\nUNKNOWN-POINTER" + (bundle / rel).write_text(yaml.safe_dump(data)) + else: + rel = "rollback.yaml" + data = yaml.safe_load((bundle / rel).read_text()) + data["removes_exactly"][-1]["name"] = "platform/workloads/keycape/sibling" + (bundle / rel).write_text(yaml.safe_dump(data)) + manifest_path = bundle / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["files"][rel] = hashlib.sha256((bundle / rel).read_bytes()).hexdigest() + manifest_path.write_text(json.dumps(manifest)) + with pytest.raises(structure.StructureRefused): + structure.prepare(CASE, plan, bundle, allow_unimplemented=True) + + +def test_missing_owner_receipt_or_generic_status_label_never_proves_materialization( + prepared, +): + request, _, _ = prepared + receipt = observations(request) + with pytest.raises(structure.StructureRefused): + structure.verify(request, receipt[:-1], lambda *args: True) + with pytest.raises(structure.StructureRefused): + structure.verify(request, receipt, lambda *args: False) + with pytest.raises(structure.StructureRefused): + structure.verify( + request, + (replace(receipt[0], disposition="secret-minted"), *receipt[1:]), + lambda *args: True, + ) + + +def test_partial_case_objects_or_exact_source_drift_refuse(prepared): + request, _, _ = prepared + receipt = observations(request) + with pytest.raises(structure.StructureRefused): + structure.verify( + request, + ( + replace(receipt[1], expected_digest="sha256:" + "0" * 64), + *receipt[:1], + *receipt[2:], + ), + lambda *args: True, + ) + + +def test_created_vs_preexisting_projection_excludes_shared_objects_from_undo(prepared): + request, _, _ = prepared + receipt = list(observations(request)) + native = [i for i, o in enumerate(request.objects) if o.proof == "native-structure"] + receipt[native[0]] = replace( + receipt[native[0]], disposition="preexisting-matched", change_receipt=None + ) + authenticated = {json.dumps(asdict(o), sort_keys=True) for o in receipt} + result = structure.verify( + request, + tuple(receipt), + lambda req, obj, obs: json.dumps(asdict(obs), sort_keys=True) in authenticated, + ) + assert request.objects[native[0]] in result.preexisting + assert request.objects[native[0]] not in result.rollback_targets + assert result.rollback_targets == tuple(reversed(result.created)) + assert ( + request.objects[0] not in result.rollback_targets + ) # layout is not secret custody undo + + +def test_source_bundle_inventory_path_escape_and_unsafe_extra_payload_refuse(prepared): + _, plan, bundle = prepared + manifest_path = bundle / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["files"]["../outside"] = "0" * 64 + manifest_path.write_text(json.dumps(manifest)) + with pytest.raises(structure.StructureRefused): + structure.prepare(CASE, plan, bundle, allow_unimplemented=True) + + +def test_owner_verifier_diagnostics_and_interruptions_are_withheld(prepared, capfd): + request, _, _ = prepared + + def broken(*args): + print("SYNTHETIC-SECRET") + os.write(2, b"SYNTHETIC-SECRET") + raise KeyboardInterrupt("SYNTHETIC-SECRET") + + with pytest.raises(structure.StructureRefused) as err: + structure.verify(request, observations(request), broken) + assert "SYNTHETIC-SECRET" not in str(err.value) + assert "SYNTHETIC-SECRET" not in "".join(capfd.readouterr()) + + +def test_full_run_guard_change_keeps_structural_request_bound_and_explicit(prepared): + _, plan, bundle = prepared + previous_digest = structure.sha(b"previous reviewed execution guard") + manifest_path = bundle / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + original = manifest["security_case"]["render_digest"] + manifest["security_case"]["render_digest"] = previous_digest + manifest_path.write_text(json.dumps(manifest)) + plan.write_text(plan.read_text().replace(original, previous_digest)) + request = structure.prepare(CASE, plan, bundle, allow_unimplemented=True) + assert request.run_source_changed is True + assert request.reviewed_render_digest == previous_digest + assert request.render_digest != previous_digest + + +def test_read_only_cli_uses_no_live_survey_or_target_write( + prepared, monkeypatch, capsys +): + from ops_mason import cli + + _, plan, bundle = prepared + monkeypatch.setattr( + case_plan, "_bao_json", lambda *args: pytest.fail("live survey must not run") + ) + before = { + path.relative_to(bundle): path.read_bytes() + for path in bundle.rglob("*") + if path.is_file() + } + result = cli.main( + [ + "structure-request", + "--case", + str(CASE), + "--plan", + str(plan), + "--bundle", + str(bundle), + "--allow-unimplemented", + ] + ) + assert result == 0 + emitted = json.loads(capsys.readouterr().out) + assert emitted["source_only"] is True + assert emitted["request"]["structural_digest"].startswith("sha256:") + assert before == { + path.relative_to(bundle): path.read_bytes() + for path in bundle.rglob("*") + if path.is_file() + } + + +def test_unknown_proof_kind_cannot_turn_credential_status_into_structure(prepared): + request, _, _ = prepared + forged = replace( + request, + objects=( + replace(request.objects[0], proof="secret-minted"), + *request.objects[1:], + ), + ) + receipt = observations(forged) + with pytest.raises(structure.StructureRefused): + structure.verify(forged, receipt, lambda *args: True) + + +def test_status_only_hash_binding_requires_real_owner_verification_and_no_mint( + prepared, +): + request, _, _ = prepared + before = CASE.read_bytes() + after_doc = yaml.safe_load(before) + after_doc["spec"]["status"] = "materialized" + after = yaml.safe_dump(after_doc).encode() + bound = structure.bind_status_transition( + request, before, after, observations(request), lambda *args: True + ) + assert bound.accepted_case_digest == request.case_digest + assert bound.materialized_case_digest == structure.sha(after) + assert bound.materialized_case_digest != bound.accepted_case_digest + assert bound.structural_digest == request.structural_digest + with pytest.raises(structure.StructureRefused): + structure.bind_status_transition( + request, before, after, observations(request), lambda *args: False + ) + + +def test_status_hash_bridge_never_carries_parameter_or_acceptance_changes(prepared): + request, _, _ = prepared + before = CASE.read_bytes() + for change in ("parameters", "acceptance"): + after = yaml.safe_load(before) + after["spec"]["status"] = "materialized" + if change == "parameters": + after["spec"]["parameters"]["client_id"] = "sibling-client" + else: + after["spec"]["acceptance"]["record"] = "new-unapproved-record" + with pytest.raises(structure.StructureRefused): + structure.bind_status_transition( + request, + before, + yaml.safe_dump(after).encode(), + observations(request), + lambda *args: True, + ) + + +def test_same_target_policy_and_role_preserve_distinct_rollback_identity(prepared): + request, _, _ = prepared + shared_target = "fixture-shared-name" + objects = tuple( + replace(obj, target=shared_target) + if obj.kind in ("openbao_policy", "openbao_role") + else obj + for obj in request.objects + ) + request = replace(request, objects=objects) + receipt = list(observations(request)) + native = [i for i, obj in enumerate(objects) if obj.proof == "native-structure"] + # One shared policy was already present; newly created policy and role must + # remain distinct even when their target names collide. + receipt[native[0]] = replace( + receipt[native[0]], disposition="preexisting-matched", change_receipt=None + ) + result = structure.verify(request, tuple(receipt), lambda *args: True) + assert objects[native[0]] in result.preexisting + assert objects[native[0]] not in result.rollback_targets + rollback = result.rollback_targets + assert {obj.kind for obj in rollback} == {"openbao_policy", "openbao_role"} + assert all(obj.target == shared_target for obj in rollback) + assert len({obj.id for obj in rollback}) == len(rollback) + assert all(obj.evidence_owner == "railiance-platform" for obj in rollback) + + +@pytest.mark.parametrize( + "field,value", + [ + ("case", "../invalid"), + ("case", []), + ("type_ref", "oidc-browser-relying-party"), + ("type_ref", "oidc-browser-relying-party@unversioned"), + ("run_source_changed", False), + ("run_source_changed", 1), + ], +) +def test_constructed_request_closed_identity_and_change_flag(prepared, field, value): + request, _, _ = prepared + forged = replace(request, **{field: value}) + with pytest.raises(structure.StructureRefused): + structure.verify(forged, observations(request), lambda *args: True) diff --git a/workplans/MASON-WP-0010-security-case-materialization.md b/workplans/MASON-WP-0010-security-case-materialization.md index 652cd0a..3fb2e46 100644 --- a/workplans/MASON-WP-0010-security-case-materialization.md +++ b/workplans/MASON-WP-0010-security-case-materialization.md @@ -9,7 +9,7 @@ flavor: implementation owner: claude-code topic_slug: netkingdom created: "2026-10-08" -updated: "2026-10-09" +updated: "2026-10-11" depends_on: [] related_repos: - net-kingdom @@ -144,7 +144,7 @@ id: MASON-WP-0010-T03 status: progress depends_on: [MASON-WP-0010-T02] blocked_on: message-from:railiance-platform -blocking_reason: "Built 2026-10-09 (builds/case-forgejo-netkingdom-oidc). Acceptance waits on railiance-platform filing the CCR drafts and the founder applying them, plus mint-secret; then the Forgejo case is live." +blocking_reason: "Typed exact structural evidence request is prepared. Native materialization waits on independently authenticated policy/role/layout/catalog observations and CCR42 consumer binding/application. Mint, delivery and first live verification are follow-through, not structural prerequisites." priority: high state_hub_task_id: "56cea6be-c613-586c-80cd-125340b374da" ``` @@ -228,3 +228,22 @@ re-pinned to @1.1. The live re-render above was already 1.1. Bundle rebuilt from it (`builds/case-forgejo-netkingdom-oidc/`, platform validation pass); the ESO policy is unchanged against the applied CCR-2026-0041, so 1.1 changes no OpenBao object. + + +2026-10-11 accepted-architecture source implementation: materialization is +structure only under the already accepted canon; the stale mint-before- +materialization blocker is removed, not replaced by a fake status label. +`src/ops_mason/case_structure.py` and read-only `ops-mason structure-request` +validate a separate current structural projection against the hashed bundle +and prepare typed per-owner observations. Full execution drift remains visible: +35880bb adds --check-config to register-client, changing full render af1f4f81 +→ e64b56e4 while materialization/composition/verification/rollback remain equal. +No prior execution approval is carried forward. Candidate receipt verification +requires independent owner/manual provenance and derives undo only from +verified creation receipts; planned rollback identities are not ownership. +Thirty-three relevant tests and ruff pass. Status-only raw-hash binding previews +accepted→materialized only after independent observations, preserving every +parameter/acceptance field; it writes and signs nothing. Details and exact native owner seams: +`docs/forgejo-structural-materialization-source-v1.md`. T03 remains progress; +case remains accepted@1.1, not materialized/live. CCR42 consumer auth remains +in_flight, and native owner evidence is not manufactured by source preparation.