diff --git a/history/2026-10-08-oidc-rp-follow-through-review.md b/history/2026-10-08-oidc-rp-follow-through-review.md index c2e03bd..75974d7 100644 --- a/history/2026-10-08-oidc-rp-follow-through-review.md +++ b/history/2026-10-08-oidc-rp-follow-through-review.md @@ -106,3 +106,14 @@ Correction to R4: the existing `rapp-qonto-keycape-client` is a client-credentials workload client at a different path. It is not a browser relying party, so it would **not** become a case of this type. Only the naming precedent carries over. + +## Re-check 2026-10-09: revised 1.0 against R1–R7 + +`warden case check-type` run on the revised type: `schema_ok=True`. R1 (`lane`), +R2 (canon definition of `warden_executes`), R3 (`value_handling`), R4′ +(templated `catalog-entry` body), R5 (`ready_check` on `deliver-secret`), R6 +(exact memberships in `ensure-groups`) and R7 (`secret-delivered`, +`mfa-pinned`) are all present. Before T02 can plan a run, two lanes must +become reviewed catalog entries: `identity-provisioner-groups` (owner +net-kingdom) and `keycape-client-secret-generate` (owner railiance-platform). +ops-warden has no further required changes to the type. diff --git a/src/warden/case_type.py b/src/warden/case_type.py new file mode 100644 index 0000000..46aae8e --- /dev/null +++ b/src/warden/case_type.py @@ -0,0 +1,170 @@ +"""SecurityCaseType follow-through check (WARDEN-WP-0044-T01). + +Checks the ``follow_through`` section of a NetKingdom SecurityCaseType against +the rules ops-warden needs before it can plan a run (canon +``security-case-types_v0.1`` section on follow_through; ops-warden review +``history/2026-10-08-oidc-rp-follow-through-review.md``, R1-R3, R5-R7). + +Two kinds of finding: + +* ``schema`` — the type itself is wrong; net-kingdom fixes the type. +* ``catalog`` — the type is right but a lane it names is not yet a reviewed, + active catalog entry; the lane's owner supplies it. ``warden case plan`` + (T02) refuses to plan while any catalog finding remains. + +Read-only: it parses YAML and the catalog, executes nothing, and never handles a +secret value. +""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable, List, Optional + +import yaml + +from warden.routing.catalog import Catalog + +VALUE_HANDLING = ("none", "generated_in_custody") +REQUIRED_STEP_FIELDS = ("id", "owner", "entry_point", "auth", "warden_executes", "attended", "value_handling") + + +@dataclass(frozen=True) +class Finding: + kind: str # "schema" | "catalog" + code: str + where: str + message: str + + def as_dict(self) -> dict: + return {"kind": self.kind, "code": self.code, "where": self.where, "message": self.message} + + +@dataclass(frozen=True) +class FollowThroughReport: + type_id: str + version: str + findings: List[Finding] + attended_lanes: List[str] # one attended session per lane, in step order + warden_steps: List[str] + + @property + def schema_ok(self) -> bool: + return not any(f.kind == "schema" for f in self.findings) + + @property + def plannable(self) -> bool: + return not self.findings + + def as_dict(self) -> dict: + return { + "type": self.type_id, + "version": self.version, + "schema_ok": self.schema_ok, + "plannable": self.plannable, + "warden_steps": self.warden_steps, + "attended_sessions": self.attended_lanes, + "findings": [f.as_dict() for f in self.findings], + } + + +def load_type(path: Path) -> dict: + with open(path) as fh: + doc = yaml.safe_load(fh) + if not isinstance(doc, dict) or doc.get("kind") != "SecurityCaseType": + raise ValueError(f"{path}: not a SecurityCaseType document") + return doc + + +def _bool(step: dict, key: str) -> Optional[bool]: + value = step.get(key) + return value if isinstance(value, bool) else None + + +def check_follow_through(doc: dict, catalog: Optional[Catalog] = None) -> FollowThroughReport: + spec = doc.get("spec") or {} + steps = spec.get("follow_through") or [] + findings: List[Finding] = [] + attended_lanes: List[str] = [] + warden_steps: List[str] = [] + + def schema(code: str, where: str, message: str) -> None: + findings.append(Finding("schema", code, where, message)) + + if not steps: + schema("FT-EMPTY", "follow_through", "type declares no follow-through steps") + + seen: set = set() + for index, step in enumerate(steps): + sid = step.get("id") if isinstance(step, dict) else None + where = f"follow_through[{sid or index}]" + if not isinstance(step, dict): + schema("FT-SHAPE", where, "step is not a mapping") + continue + for key in REQUIRED_STEP_FIELDS: + if key not in step or step[key] in (None, ""): + schema("FT-FIELD", where, f"missing required field '{key}'") + if sid in seen: + schema("FT-DUP", where, "duplicate step id") + seen.add(sid) + + executes = _bool(step, "warden_executes") + attended = _bool(step, "attended") + for key, value in (("warden_executes", executes), ("attended", attended)): + if key in step and value is None: + schema("FT-BOOL", where, f"'{key}' must be a boolean") + + # R3: values never pass through the run. + handling = step.get("value_handling") + if handling is not None and handling not in VALUE_HANDLING: + schema("FT-VALUE", where, f"value_handling '{handling}' not in {list(VALUE_HANDLING)}") + if handling == "generated_in_custody" and attended: + schema("FT-VALUE", where, "a generated-in-custody step may not run in an attended session") + + # Attended steps: justified, and never executed by warden itself. + if attended: + if not step.get("attended_justification"): + schema("FT-ATTEND", where, "attended step carries no attended_justification") + if executes: + schema("FT-ATTEND", where, "a step is either warden_executes or attended, not both") + + # R1: a lane for every step warden executes or groups into a session. + lane = step.get("lane") + if (executes or attended) and not lane: + schema("FT-LANE", where, "warden_executes or attended step names no catalog lane") + if lane and catalog is not None: + entry = catalog.get(lane) + if entry is None: + findings.append(Finding("catalog", "LANE-MISSING", where, f"lane '{lane}' is not in the catalog")) + elif not entry.is_active or not entry.reviewed: + findings.append(Finding("catalog", "LANE-UNREVIEWED", where, f"lane '{lane}' is not active and reviewed")) + elif attended and entry.lane != "login": + findings.append(Finding("catalog", "LANE-KIND", where, f"attended step's lane '{lane}' is not a login lane")) + + if executes: + warden_steps.append(sid) + if attended and lane and lane not in attended_lanes: + attended_lanes.append(lane) + + # R7: verification never reads a value. + for index, item in enumerate(spec.get("verification") or []): + if not isinstance(item, dict) or item.get("reads_secret_value") is not False: + vid = item.get("id", index) if isinstance(item, dict) else index + schema("VER-VALUE", f"verification[{vid}]", "reads_secret_value must be explicitly false") + + # Every step must be reversible (or explicitly the consumer's to undo). + reversed_ids = {r.get("reverses") for r in spec.get("rollback") or [] if isinstance(r, dict)} + for sid in sorted(s for s in seen if s and s not in reversed_ids): + schema("RB-MISSING", f"follow_through[{sid}]", "no rollback entry reverses this step") + + return FollowThroughReport( + type_id=str(spec.get("id", "")), + version=str(spec.get("version", "")), + findings=findings, + attended_lanes=attended_lanes, + warden_steps=warden_steps, + ) + + +def summarize(findings: Iterable[Finding]) -> str: + return "\n".join(f"[{f.kind}] {f.code} {f.where}: {f.message}" for f in findings) diff --git a/src/warden/cli.py b/src/warden/cli.py index 6c2d0be..80fd467 100644 --- a/src/warden/cli.py +++ b/src/warden/cli.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Annotated, List, Optional import typer +import yaml from rich.console import Console from rich.table import Table @@ -50,6 +51,11 @@ memory_app = typer.Typer( no_args_is_help=True, ) app.add_typer(memory_app, name="memory") +case_app = typer.Typer( + help="SecurityCaseType follow-through (WARDEN-WP-0044; read-only check, no execution)", + no_args_is_help=True, +) +app.add_typer(case_app, name="case") console = Console() err = Console(stderr=True) @@ -2212,3 +2218,28 @@ def memory_activate( print(json.dumps(payload, indent=2)) return console.print(warden_memory.format_activation_summary(payload)) + + +@case_app.command("check-type") +def case_check_type( + type_file: Annotated[Path, typer.Argument(help="SecurityCaseType type.yaml")], + output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False, +) -> None: + """Check a type's follow_through against ops-warden's planning rules (exit 1 if not plannable).""" + from warden.case_type import check_follow_through, load_type, summarize + + try: + doc = load_type(type_file) + except (OSError, ValueError, yaml.YAMLError) as exc: + err.print(f"[red]{exc}[/red]") + raise typer.Exit(2) + report = check_follow_through(doc, _load_catalog()) + if output_json: + print(json.dumps(report.as_dict(), indent=2)) + else: + console.print(f"{report.type_id} {report.version}: schema_ok={report.schema_ok} plannable={report.plannable}") + console.print(f"warden executes: {', '.join(report.warden_steps) or '-'}") + console.print(f"attended sessions: {', '.join(report.attended_lanes) or '-'}") + if report.findings: + console.print(summarize(report.findings), markup=False) + raise typer.Exit(0 if report.plannable else 1) diff --git a/tests/test_case_type.py b/tests/test_case_type.py new file mode 100644 index 0000000..bec2ed6 --- /dev/null +++ b/tests/test_case_type.py @@ -0,0 +1,127 @@ +"""WARDEN-WP-0044-T01: SecurityCaseType follow-through check.""" +from __future__ import annotations + +import copy +from pathlib import Path + +import pytest +import yaml +from typer.testing import CliRunner + +from warden.case_type import check_follow_through, load_type +from warden.cli import app +from warden.routing.catalog import load_catalog + +LIVE_TYPE = Path.home() / "net-kingdom/security-case-types/oidc-browser-relying-party/type.yaml" + +BASE = { + "apiVersion": "netkingdom.io/security-case-type/v0.1", + "kind": "SecurityCaseType", + "spec": { + "id": "t", + "version": "1.0", + "follow_through": [ + {"id": "run", "owner": "o", "entry_point": "e", "auth": "a", "lane": "key-cape-oidc-login", + "warden_executes": True, "attended": False, "value_handling": "none"}, + {"id": "login", "owner": "o", "entry_point": "e", "auth": "a", "lane": "key-cape-oidc-login", + "warden_executes": False, "attended": True, "value_handling": "none", + "attended_justification": "live registry is a Secret"}, + {"id": "handover", "owner": "o", "entry_point": "e", "auth": "a", + "warden_executes": False, "attended": False, "value_handling": "none"}, + ], + "verification": [{"id": "v", "check": "c", "expect": "present", "reads_secret_value": False}], + "rollback": [{"reverses": "run"}, {"reverses": "login"}, {"reverses": "handover"}], + }, +} + + +@pytest.fixture(scope="module") +def catalog(): + return load_catalog() + + +def _doc(**step_overrides): + doc = copy.deepcopy(BASE) + doc["spec"]["follow_through"][0].update(step_overrides) + return doc + + +def _codes(report): + return {f.code for f in report.findings} + + +def test_conforming_type_is_plannable(catalog): + report = check_follow_through(_doc(), catalog) + assert report.plannable, report.findings + assert report.warden_steps == ["run"] + assert report.attended_lanes == ["key-cape-oidc-login"] + + +def test_executed_step_without_lane_refused(catalog): + doc = _doc() + del doc["spec"]["follow_through"][0]["lane"] + assert "FT-LANE" in _codes(check_follow_through(doc, catalog)) + + +def test_unknown_lane_is_catalog_finding_not_schema(catalog): + report = check_follow_through(_doc(lane="no-such-lane"), catalog) + assert report.schema_ok and not report.plannable + assert _codes(report) == {"LANE-MISSING"} + + +def test_value_handling_required_and_enumerated(catalog): + assert "FT-VALUE" in _codes(check_follow_through(_doc(value_handling="returned"), catalog)) + doc = _doc() + del doc["spec"]["follow_through"][0]["value_handling"] + assert "FT-FIELD" in _codes(check_follow_through(doc, catalog)) + + +def test_attended_needs_justification_and_excludes_execution(catalog): + doc = copy.deepcopy(BASE) + del doc["spec"]["follow_through"][1]["attended_justification"] + doc["spec"]["follow_through"][1]["warden_executes"] = True + msgs = [f.message for f in check_follow_through(doc, catalog).findings if f.code == "FT-ATTEND"] + assert len(msgs) == 2 + + +def test_attended_step_needs_login_lane(catalog): + doc = copy.deepcopy(BASE) + secret_lane = next(e.id for e in catalog.entries if e.is_active and e.reviewed and e.lane == "secret") + doc["spec"]["follow_through"][1]["lane"] = secret_lane + assert "LANE-KIND" in _codes(check_follow_through(doc, catalog)) + + +def test_generated_value_never_in_attended_session(catalog): + doc = copy.deepcopy(BASE) + doc["spec"]["follow_through"][1]["value_handling"] = "generated_in_custody" + assert "FT-VALUE" in _codes(check_follow_through(doc, catalog)) + + +def test_verification_must_not_read_values(catalog): + doc = _doc() + doc["spec"]["verification"][0]["reads_secret_value"] = True + assert "VER-VALUE" in _codes(check_follow_through(doc, catalog)) + + +def test_every_step_reversible(catalog): + doc = _doc() + doc["spec"]["rollback"] = doc["spec"]["rollback"][1:] + assert "RB-MISSING" in _codes(check_follow_through(doc, catalog)) + + +def test_cli_exit_codes(tmp_path): + good, bad = tmp_path / "good.yaml", tmp_path / "bad.yaml" + good.write_text(yaml.safe_dump(_doc())) + bad.write_text(yaml.safe_dump(_doc(lane="no-such-lane"))) + runner = CliRunner() + assert runner.invoke(app, ["case", "check-type", str(good), "--json"]).exit_code == 0 + assert runner.invoke(app, ["case", "check-type", str(bad), "--json"]).exit_code == 1 + + +@pytest.mark.skipif(not LIVE_TYPE.exists(), reason="net-kingdom checkout not present") +def test_live_oidc_rp_type_schema_conforms(catalog): + """The revised 1.0 type meets R1-R3/R5-R7; only the two owner lanes are outstanding.""" + report = check_follow_through(load_type(LIVE_TYPE), catalog) + assert report.schema_ok, report.findings + assert report.attended_lanes == ["key-cape-oidc-login"] + assert {f.code for f in report.findings} <= {"LANE-MISSING", "LANE-UNREVIEWED"} diff --git a/workplans/WARDEN-WP-0044-security-case-follow-through.md b/workplans/WARDEN-WP-0044-security-case-follow-through.md index e3c55d2..113ec69 100644 --- a/workplans/WARDEN-WP-0044-security-case-follow-through.md +++ b/workplans/WARDEN-WP-0044-security-case-follow-through.md @@ -9,7 +9,7 @@ flavor: implementation owner: claude-code topic_slug: netkingdom created: "2026-10-08" -updated: "2026-10-08" +updated: "2026-10-09" depends_on: [] related_repos: - net-kingdom @@ -66,6 +66,15 @@ net-kingdom acknowledges the review (NK-WP-0048-T05). `{consuming_service}-keycape-client`, with the entry body templated in the type (R4′, see the addendum in the review). +2026-10-09: the revised 1.0 type and canon now carry R1–R3, R4′, R5–R7. +`warden case check-type ` (`src/warden/case_type.py`, tests in +`tests/test_case_type.py`) checks a type's follow-through mechanically: the +live type is schema-conformant, with one attended session +(`key-cape-oidc-login`). It is not yet plannable, because two lanes are not +in the catalog: `identity-provisioner-groups` (net-kingdom) and +`keycape-client-secret-generate` (railiance-platform). These gate T02, not +T01. T01 still closes on net-kingdom's acknowledgement (NK-WP-0048-T05). + ```task id: WARDEN-WP-0044-T02 status: wait