"""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) # The committed historical bundle mislabels its consumer target keycape. # Preserve it in the owner checkout; corrected fixture metadata must match # the current generator rather than weakening the exact structural checker. rel = "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml" ccr = yaml.safe_load((bundle / rel).read_text()) ccr["target"]["workload"] = "forgejo" (bundle / rel).write_text(yaml.safe_dump(ccr, sort_keys=False, width=110)) manifest = json.loads((bundle / "manifest.json").read_text()) manifest["files"][rel] = hashlib.sha256((bundle / rel).read_bytes()).hexdigest() (bundle / "manifest.json").write_text(json.dumps(manifest, indent=1, sort_keys=True)+"\n") 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_corrected_first_case_fixture_structure_and_full_execution_drift_remain( 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) def test_historical_bundle_wrong_consumer_target_is_not_grandfathered(tmp_path): bundle = tmp_path / "historical-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) assert yaml.safe_load((bundle / "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml").read_text())["target"]["workload"] == "keycape" with pytest.raises(structure.StructureRefused, match="role or reader binding differs"): structure.prepare(CASE, plan, bundle, allow_unimplemented=True)