"""Integration: the real net-kingdom 1.3 candidate against this kernel and catalog. Reads the actual 1.3 type, its forgejo example and the committed composition artifact from net-kingdom, plus this branch's real routing catalog. The descriptor map and metadata come from those committed files. All authority is fixture-only: we change the proposed type to acknowledged, change the example case to materialized and replace its acceptance record, promote draft lanes in memory, and temporarily install the 1.3 parameter pin. Stub adapters replace native adapters; an unconditional evidence verifier simulates every acceptance and caller session. WARDEN_AGENT_ID is removed to simulate a human runtime. No adapter is dispatched. This proves conditional source admission, not authenticated owner acceptance, native execution or readiness; unresolved Apps sessions stay refused. """ from dataclasses import asdict import hashlib import json import os from pathlib import Path from unittest import mock import pytest import yaml from warden import case_inputs as inputs from warden.case_receipt_state import Refused from warden.case_run import Adapter, Descriptor, Executor, Journal, digest from warden.case_type import check_follow_through from warden.routing.catalog import load_catalog NK = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT", Path(__file__).resolve().parents[2] / "net-kingdom")) TYPE_DIR = NK / "security-case-type-revisions/1.3/oidc-browser-relying-party" ARTIFACT = NK / "sso-mfa/k8s/keycape/descriptor-composition-1.3.candidate.json" CATALOG = Path(__file__).resolve().parents[1] / "registry/routing/catalog.yaml" pytestmark = pytest.mark.skipif(not (TYPE_DIR / "type.yaml").is_file() or not ARTIFACT.is_file(), reason="net-kingdom 1.3 candidate not available") DRAFT_LANES = {"keycape-attended-case-change", "platform-secret-delivery", "consumer-owner-deployment", "apps-consumer-handover"} def encode(desc): # net-kingdom compose-1.2-descriptors.py encode() return json.dumps(asdict(desc), sort_keys=True, separators=(",", ":")).encode() def type_doc(status="acknowledged"): doc = yaml.safe_load((TYPE_DIR / "type.yaml").read_bytes()) doc["spec"]["status"] = status return doc def catalog_bytes(promote): raw = yaml.safe_load(CATALOG.read_bytes()) if promote: for entry in raw["entries"]: if entry["id"] in DRAFT_LANES and entry.get("status") == "draft": entry["status"] = "active" return yaml.safe_dump(raw).encode() def descriptors(): art = json.loads(ARTIFACT.read_text()) out = {} for row in art["steps"]: d = dict(row["descriptor"]) for key in ("allowed_created", "revocation_scopes"): d[key] = tuple(d[key]) out[row["step"]] = Descriptor(**d) return out def pin_13(): params = type_doc()["spec"]["parameters"] fp = hashlib.sha256(json.dumps(params, sort_keys=True, separators=(",", ":")).encode()).hexdigest() return mock.patch.dict(inputs.PARAMETER_PINS, {"1.3": fp}) def unused(*_): raise AssertionError("admission must not dispatch") def admit(tmp_path, desc, *, promote=True, pinned=True, status="acknowledged"): case = yaml.safe_load((TYPE_DIR / "examples/forgejo.yaml").read_bytes()) case["spec"].update(status="materialized", acceptance={ "at": "2026-10-17", "record": "fixture-only acceptance, not an owner record", "by": [{"name": "fixture", "role": "owner"}, {"name": "fixture", "role": "founder"}]}) adapter = Adapter(desc.adapter, desc.adapter_source, unused, unused) executor = Executor(Journal(tmp_path / desc.step), {adapter.reference: adapter}, lambda purpose, source, evidence: True, browser_inputs=inputs.BrowserInputs(NK)) with mock.patch.dict(os.environ), (pin_13() if pinned else mock.patch.dict({})): os.environ.pop("WARDEN_AGENT_ID", None) return executor.admit(yaml.safe_dump(case).encode(), yaml.safe_dump(type_doc(status)).encode(), encode(desc), catalog_bytes(promote), principal="founder", actor=desc.actor, lane=desc.lane, acceptance_evidence="e-acceptance", materialization_evidence="e-mason", descriptor_evidence="e-descriptor", session_evidence="e-session") def test_artifact_descriptors_are_the_type_map(): descs = descriptors() assert len(descs) == 7 assert {s: digest(encode(d)) for s, d in descs.items()} == type_doc()["spec"]["execution_descriptors"] def test_conditional_admission_keeps_unresolved_apps_sessions_refused(tmp_path): for step, desc in descriptors().items(): if step in ("deliver-consumer-secret", "hand-over"): with pytest.raises(Refused, match="does not resolve to an active, reviewed login lane"): admit(tmp_path, desc) else: admitted, invocation, _ = admit(tmp_path, desc) assert admitted == desc assert (invocation.step, invocation.actor, invocation.target) == (step, "human", "forgejo") def test_draft_lane_steps_refused_until_promoted(tmp_path): for step, desc in descriptors().items(): if desc.lane in DRAFT_LANES: with pytest.raises(Refused, match="descriptor lane is not active in the catalog"): admit(tmp_path, desc, promote=False) else: admit(tmp_path, desc, promote=False) # mint-secret's lane is already active def test_unpinned_1_3_refused_by_browser_validation(tmp_path): with pytest.raises(Refused, match="failed exact-version validation"): admit(tmp_path, descriptors()["mint-secret"], pinned=False) def test_proposed_type_refused(tmp_path): with pytest.raises(Refused, match="type is not acknowledged"): admit(tmp_path, descriptors()["mint-secret"], status="proposed") def test_type_check_findings_against_the_real_catalog(tmp_path): def findings(promote): path = tmp_path / f"catalog-{promote}.yaml" path.write_bytes(catalog_bytes(promote)) report = check_follow_through(type_doc("proposed"), load_catalog(path)) return {(f.code, f.where) for f in report.findings} draft = findings(False) assert {code for code, _ in draft} == {"LANE-UNREVIEWED"} assert len(draft) == 6 # six steps use the four draft lanes; mint-secret's lane is active promoted = findings(True) # Only the two railiance-apps lanes remain: their session_lane awaits owner confirmation. assert {code for code, _ in promoted} == {"LANE-KIND"} assert len(promoted) == 2 assert all("deliver-consumer-secret" in where or "hand-over" in where for _, where in promoted)