Prepare reviewed Forgejo consumer ESO scope and truthful recovery records

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_case_build.py

@@ -109,3 +109,64 @@ def test_platform_validator_accepts_the_drafts(tmp_path, types_root):
     r = cb.build_from_case(case, _plan(tmp_path, case, types_root), tmp_path / "b", types_root=types_root,
                            audit_log=tmp_path / "a.jsonl")
     assert cb.validate_with_platform([Path(p) for p in r["ccrs"]]) == []
+
+
+def test_two_role_backed_readers_keep_distinct_ccrs_and_complete_structure(tmp_path, types_root):
+    from ops_mason import case_structure
+    type_path = types_root / TYPE.name / "type.yaml"
+    doc = yaml.safe_load(type_path.read_text())
+    doc["spec"]["materialization"].insert(-1, {
+        "id": "consumer-eso-role", "kind": "openbao_role",
+        "when": "client_type == confidential",
+        "template": "external-secrets-keycape-{client_id}-{consuming_service}",
+        "body": "auth: kubernetes\nbound_service_account_names: [forgejo-keycape-forgejo-eso]\nbound_service_account_namespaces: [forgejo]\ntoken_policies: [workload-kv-read-keycape-{client_id}-{consuming_service}]\ntoken_no_default_policy: true\ntoken_ttl: 15m\ntoken_max_ttl: 15m\n",
+    })
+    doc["spec"]["rollback"].append({"reverses": "consumer-eso-role", "when": "client_type == confidential", "action": "remove this created role only"})
+    type_path.write_text(yaml.safe_dump(doc, sort_keys=False))
+    case = _case(tmp_path)
+    plan = _plan(tmp_path, case, types_root)
+    out = tmp_path / "bundle"
+    result = cb.build_from_case(case, plan, out, types_root=types_root, audit_log=tmp_path / "a.jsonl")
+    assert len(result["ccrs"]) == len(set(result["ccrs"])) == 2
+    ccrs = {Path(p).name: yaml.safe_load(Path(p).read_text()) for p in result["ccrs"]}
+    assert ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo"
+    consumer = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"]
+    assert ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["target"]["workload"] == "keycape"
+    assert consumer["target"]["workload"] == "forgejo"
+    assert consumer["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo-forgejo"
+    assert consumer["openbao"]["auth"]["bound_claims"] == {"service_account_names": ["forgejo-keycape-forgejo-eso"], "service_account_namespaces": ["forgejo"]}
+    assert not consumer["openbao"]["auth"]["bound_claims_confirmed"]
+    deactivate = consumer["lifecycle"]["deactivate"].lower()
+    for guard in ["created", "preexisting", "consumer reference", "revoke issued", "verified expiry", "no new login", "no remint"]:
+        assert guard in deactivate
+    assert "delete the role, then policy" in ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]["lifecycle"]["deactivate"]
+    assert cb.validate_with_platform([Path(p) for p in result["ccrs"]]) == []
+    request = case_structure.prepare(case, plan, out, types_root=types_root)
+    assert len(request.objects) == 6
+    assert {o.id for o in request.objects} >= {"eso-role", "consumer-eso-role"}
+
+
+def test_duplicate_source_output_refuses_before_overwriting_first_ccr(tmp_path, types_root, monkeypatch):
+    original = cb._ccr
+    def same_identity(*args, **kwargs):
+        doc = original(*args, **kwargs)
+        doc["id"] = "CCR-DRAFT-COLLISION"
+        return doc
+    monkeypatch.setattr(cb, "_ccr", same_identity)
+    case = _case(tmp_path)
+    out = tmp_path / "bundle"
+    with pytest.raises(cb.CaseBuildRefused, match="duplicate output identity"):
+        cb.build_from_case(case, _plan(tmp_path, case, types_root), out, types_root=types_root, audit_log=tmp_path / "a.jsonl")
+    first = yaml.safe_load((out / "credential-change-requests/CCR-DRAFT-COLLISION.yaml").read_text())
+    assert first["openbao"]["policy_name"] == "workload-kv-read-keycape-forgejo-eso"
+    assert not (out / "manifest.json").exists()
+
+
+def test_generated_consumer_lifecycle_requires_created_only_token_recovery(types_root):
+    case = yaml.safe_load(CASE.read_text())
+    policy = {"id": "consumer-reader-policy", "template": "workload-kv-read-keycape-forgejo-forgejo"}
+    role = {"template": "external-secrets-keycape-forgejo-forgejo", "body": "auth: kubernetes\nbound_service_account_names: [forgejo-keycape-forgejo-eso]\nbound_service_account_namespaces: [forgejo]\ntoken_policies: [workload-kv-read-keycape-forgejo-forgejo]\n"}
+    provenance = {"case": "forgejo-netkingdom-oidc", "type": "fixture-source-only", "render_digest": "fixture", "accepted_by": ["fixture-owner"], "accepted_at": "fixture-date", "acceptance_record": "fixture-not-authority"}
+    doc = cb._ccr(case, {}, policy, role, "platform/workloads/keycape/forgejo", provenance)
+    for guard in ["created", "preexisting", "consumer reference", "revoke issued", "verified expiry", "no new login", "no remint"]:
+        assert guard in doc["lifecycle"]["deactivate"].lower()

tests/test_case_structure.py

@@ -20,6 +20,16 @@ def prepared(tmp_path):
     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
 
@@ -42,7 +52,7 @@ def observations(request):
     return tuple(result)
 
 
-def test_actual_first_case_structure_matches_but_full_execution_guard_drift_remains(
+def test_corrected_first_case_fixture_structure_and_full_execution_drift_remain(
     prepared,
 ):
     request, _, _ = prepared
@@ -322,3 +332,13 @@ def test_constructed_request_closed_identity_and_change_flag(prepared, field, va
     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)