diff --git a/src/warden/cli.py b/src/warden/cli.py index a7fdd46..6c2d0be 100644 --- a/src/warden/cli.py +++ b/src/warden/cli.py @@ -702,6 +702,26 @@ def _print_entry_table( console.print(table) +@route_app.command("scaffold") +def route_scaffold( + input_path: Annotated[Path, typer.Option("--input", help="Non-secret draft request YAML")], + output_dir: Annotated[Path, typer.Option("--output-dir", help="New staging directory")], +) -> None: + """Generate validated draft catalog/playbook pointers offline.""" + from warden.routing.scaffold import ScaffoldError, scaffold + from warden.routing.catalog import CatalogError + + try: + paths = scaffold(input_path, output_dir) + except ScaffoldError as exc: + err.print(f"[red]Scaffold refused:[/red] {exc}") + raise typer.Exit(1) + except CatalogError: + err.print("[red]Scaffold refused:[/red] existing routing catalog is invalid") + raise typer.Exit(1) + console.print("Draft staging created: " + ", ".join(path.name for path in paths)) + + @route_app.command("list") def route_list( output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False, diff --git a/src/warden/routing/scaffold.py b/src/warden/routing/scaffold.py new file mode 100644 index 0000000..c82ec38 --- /dev/null +++ b/src/warden/routing/scaffold.py @@ -0,0 +1,198 @@ +"""Offline, draft-only catalog scaffolding (WARDEN-WP-0041).""" + +from __future__ import annotations + +from copy import deepcopy +from datetime import date +from pathlib import Path +import re +import shutil + +import yaml + +from warden.routing.catalog import ( + Catalog, + CatalogError, + _assert_no_secret_material, + _parse_entry, + load_catalog, +) + + +class ScaffoldError(ValueError): + """Safe diagnostic: field names only, never supplied values.""" + + +FIELDS = frozenset( + { + "schema_version", + "lane_id", + "title", + "need_keywords", + "consumer_repo", + "owner_repo", + "workload_ref", + "risk", + "wiki_ref", + "canon_ref", + "platform_request_ref", + "reviewed", + } +) +WORKLOAD_FIELDS = frozenset( + { + "applicability", + "rapp_id", + "name", + "deployable", + "declaration_ref", + "reason", + "unknown_reason", + } +) +SLUG = re.compile(r"[a-z][a-z0-9]*(?:-[a-z0-9]+)*\Z") + + +def _screen(value: object, field: str) -> None: + if isinstance(value, dict): + for key, item in value.items(): + _screen(item, field) + elif isinstance(value, list): + for item in value: + _screen(item, field) + elif isinstance(value, str): + try: + _assert_no_secret_material("draft", field, value, prose=False) + except CatalogError: + raise ScaffoldError(f"{field}: suspected secret material") from None + + +def render(request: object, catalog: Catalog) -> dict[str, str]: + """Validate before any write; preserve explicit owner metadata.""" + if not isinstance(request, dict) or set(request) != FIELDS: + raise ScaffoldError( + "request: required fields missing or unknown fields present" + ) + if type(request["schema_version"]) is not int or request["schema_version"] != 1: + raise ScaffoldError("schema_version: expected 1") + for field in FIELDS - {"schema_version", "need_keywords", "workload_ref"}: + value = request[field] + if ( + not isinstance(value, str) + or not value.strip() + or any(c in value for c in "\r\n\x00") + ): + raise ScaffoldError(f"{field}: expected nonempty single-line text") + _screen(value, field) + for field in ("lane_id", "consumer_repo", "owner_repo"): + if not SLUG.fullmatch(request[field]): + raise ScaffoldError(f"{field}: expected repository/lane slug") + lane = request["lane_id"] + if catalog.get(lane) is not None: + raise ScaffoldError("lane_id: already exists in routing catalog") + if request["owner_repo"] != "railiance-platform": + raise ScaffoldError( + "owner_repo: this scaffold supports railiance-platform KV read pointers" + ) + if request["risk"] not in ("standard", "high", "ungraded"): + raise ScaffoldError("risk: expected explicit standard, high or ungraded") + if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", request["reviewed"]): + raise ScaffoldError("reviewed: expected ISO date") + try: + date.fromisoformat(request["reviewed"]) + except ValueError: + raise ScaffoldError("reviewed: expected ISO date") from None + keywords = request["need_keywords"] + if ( + not isinstance(keywords, list) + or not keywords + or any( + not isinstance(k, str) or not k.strip() or any(c in k for c in "\r\n\x00") + for k in keywords + ) + ): + raise ScaffoldError("need_keywords: expected nonempty text list") + _screen(keywords, "need_keywords") + workload = request["workload_ref"] + if not isinstance(workload, dict) or not set(workload) <= WORKLOAD_FIELDS: + raise ScaffoldError("workload_ref: expected known catalog fields") + if any( + not isinstance(v, str) or not v.strip() or any(c in v for c in "\r\n\x00") + for v in workload.values() + ): + raise ScaffoldError("workload_ref: expected nonempty text values") + _screen(workload, "workload_ref") + entry = { + "id": lane, + "title": request["title"], + "need_keywords": deepcopy(keywords), + "owner_repo": request["owner_repo"], + "subsystem": "openbao", + "warden_executes": False, + "wiki_ref": request["wiki_ref"], + "canon_ref": request["canon_ref"], + "reviewed": request["reviewed"], + "status": "draft", + "workload_ref": deepcopy(workload), + "risk": request["risk"], + "lane": "secret", + "exec_capable": False, + } + try: + _parse_entry(entry, 0) + except CatalogError: + raise ScaffoldError( + "workload_ref: invalid catalog workload declaration" + ) from None + playbook = f"""# {request["title"]} + +Draft lane: `{lane}`. Consumer: `{request["consumer_repo"]}`. +Owner: `{request["owner_repo"]}`. +Platform request: `{request["platform_request_ref"]}`. +Authoritative reference: `{request["canon_ref"]}`. + +## Worker checklist + +- Obtain platform review and confirmed workload/auth bindings. +- Review delegated frontdoor and whole-path rotation requirements with owners. +- Use the platform owner's CCR lifecycle and evidence procedure. +- Record approved metadata, attended values and deployment receipts separately. +- Obtain positive/negative capability evidence without reading secret data. +- Review catalog-lane-promotion criteria before any activation. + +This draft supplies no executable fetch, approval, grant or runtime readiness. +""" + return { + "catalog.yaml": yaml.safe_dump({"entries": [entry]}, sort_keys=False), + "playbook.md": playbook, + "request.yaml": yaml.safe_dump(deepcopy(request), sort_keys=True), + } + + +def scaffold( + input_path: Path, output_dir: Path, *, catalog: Catalog | None = None +) -> list[Path]: + """Publish a new staging directory; never replace existing destinations.""" + try: + request = yaml.safe_load(input_path.read_text()) + except (OSError, yaml.YAMLError): + raise ScaffoldError("input: unreadable or invalid YAML") from None + outputs = render(request, catalog if catalog is not None else load_catalog()) + if output_dir.exists() or output_dir.is_symlink(): + raise ScaffoldError("output_dir: destination already exists") + if not output_dir.parent.is_dir() or output_dir.parent.is_symlink(): + raise ScaffoldError( + "output_dir: parent must be an existing nonsymlink directory" + ) + try: + output_dir.mkdir() # exclusive creation; collision races fail before writes + except OSError: + raise ScaffoldError("output_dir: cannot create new destination") from None + try: + for name, content in outputs.items(): + with (output_dir / name).open("x") as stream: + stream.write(content) + except OSError: + shutil.rmtree(output_dir) + raise ScaffoldError("output_dir: publication failed") from None + return [output_dir / name for name in outputs] diff --git a/tests/test_scaffold.py b/tests/test_scaffold.py new file mode 100644 index 0000000..ad1c2bd --- /dev/null +++ b/tests/test_scaffold.py @@ -0,0 +1,193 @@ +from copy import deepcopy +import socket +import subprocess + +import pytest +import yaml +from typer.testing import CliRunner + +from warden.cli import app +from warden.routing.catalog import Catalog, load_catalog +from warden.routing.scaffold import ScaffoldError, render, scaffold + + +@pytest.fixture +def draft_request(): + return { + "schema_version": 1, + "lane_id": "example-workload-kv", + "title": "Example workload read", + "need_keywords": ["example workload"], + "consumer_repo": "example-consumer", + "owner_repo": "railiance-platform", + "workload_ref": { + "applicability": "applicable", + "unknown_reason": "Synthetic fixture has no published deployment identity", + }, + "risk": "standard", + "wiki_ref": "railiance-platform:docs/catalog-lane-scaffold.md", + "canon_ref": "railiance-platform:credential-change-requests/CCR-2026-9999.yaml", + "platform_request_ref": "railiance-platform:examples/catalog-lane-scaffold/platform-request.yaml", + "reviewed": "2026-10-04", + } + + +def empty_catalog(tmp_path): + return Catalog(path=tmp_path / "unused.yaml", entries=[]) + + +def test_deterministic_draft_roundtrip_and_input_unchanged(draft_request, tmp_path): + before = deepcopy(draft_request) + contents = render(draft_request, empty_catalog(tmp_path)) + assert contents == render(draft_request, empty_catalog(tmp_path)) + assert draft_request == before + for name in ["one", "two"]: + source = tmp_path / f"{name}.yaml" + source.write_text(yaml.safe_dump(draft_request)) + paths = scaffold(source, tmp_path / name, catalog=empty_catalog(tmp_path)) + assert len(paths) == 3 + assert (tmp_path / "one/catalog.yaml").read_bytes() == ( + tmp_path / "two/catalog.yaml" + ).read_bytes() + entry = load_catalog(tmp_path / "one/catalog.yaml").entries[0] + assert entry.status == "draft" + assert not entry.exec_capable and not entry.warden_executes + assert entry.fetch_command is None + assert entry.workload_ref.resolution == "unknown" + assert entry.workload_ref.name is None + assert entry.risk == "standard" + assert yaml.safe_load((tmp_path / "one/request.yaml").read_text()) == draft_request + + +@pytest.mark.parametrize("risk", ["high", "ungraded"]) +def test_explicit_risk_is_preserved(draft_request, tmp_path, risk): + draft_request["risk"] = risk + raw = yaml.safe_load(render(draft_request, empty_catalog(tmp_path))["catalog.yaml"]) + assert raw["entries"][0]["risk"] == risk + + +@pytest.mark.parametrize( + "mutation", + [ + lambda r: r.update(exec_capable=True), + lambda r: r.pop("risk"), + lambda r: r.update(risk=""), + lambda r: r.update(schema_version=True), + lambda r: r.update(lane_id="../escape"), + lambda r: r.update(reviewed="yesterday"), + lambda r: r.update(need_keywords=[]), + lambda r: r.update(workload_ref={"applicability": "applicable"}), + lambda r: r.update( + workload_ref={ + "applicability": "applicable", + "unknown_reason": "unknown", + "name": "inferred", + } + ), + lambda r: r.update( + workload_ref={ + "applicability": "applicable", + "unknown_reason": "unknown", + "role": "admin", + } + ), + ], +) +def test_invalid_request_publishes_nothing(draft_request, tmp_path, mutation): + mutation(draft_request) + source = tmp_path / "input.yaml" + source.write_text(yaml.safe_dump(draft_request)) + with pytest.raises(ScaffoldError): + scaffold(source, tmp_path / "stage", catalog=empty_catalog(tmp_path)) + assert not (tmp_path / "stage").exists() + + +@pytest.mark.parametrize( + "field", ["title", "platform_request_ref", "need_keywords", "workload_ref"] +) +def test_secret_rejection_does_not_echo_value(draft_request, tmp_path, field): + token = "ghp_" + "Ab12" * 12 + draft_request[field] = ( + [token] + if field == "need_keywords" + else {"applicability": "applicable", "unknown_reason": token} + if field == "workload_ref" + else token + ) + with pytest.raises(ScaffoldError) as exc: + render(draft_request, empty_catalog(tmp_path)) + assert token not in str(exc.value) + assert "Ab12" not in str(exc.value) + + +def test_existing_lane_collision_even_when_draft(draft_request, tmp_path): + raw = render(draft_request, empty_catalog(tmp_path))["catalog.yaml"] + path = tmp_path / "catalog.yaml" + path.write_text(raw) + with pytest.raises(ScaffoldError, match="already exists"): + render(draft_request, load_catalog(path)) + + +@pytest.mark.parametrize("symlink", [False, True]) +def test_preexisting_destination_is_preserved(draft_request, tmp_path, symlink): + source = tmp_path / "input.yaml" + source.write_text(yaml.safe_dump(draft_request)) + stage = tmp_path / "stage" + if symlink: + stage.symlink_to(tmp_path / "missing") + else: + stage.mkdir() + (stage / "keep").write_text("original") + with pytest.raises(ScaffoldError, match="already exists"): + scaffold(source, stage, catalog=empty_catalog(tmp_path)) + if not symlink: + assert (stage / "keep").read_text() == "original" + else: + assert stage.is_symlink() + + +def test_cli_offline_and_bad_yaml_diagnostic(draft_request, tmp_path, monkeypatch): + def forbidden(*args, **kwargs): + raise AssertionError("credential/network execution forbidden") + + monkeypatch.setattr(socket, "create_connection", forbidden) + monkeypatch.setattr(subprocess, "run", forbidden) + monkeypatch.setenv("WARDEN_MEMORY", "0") + source = tmp_path / "input.yaml" + source.write_text(yaml.safe_dump(draft_request)) + # Override live catalog with a different inert draft lane. + other = deepcopy(draft_request) + other["lane_id"] = "other-example" + catalog_path = tmp_path / "catalog.yaml" + catalog_path.write_text(render(other, empty_catalog(tmp_path))["catalog.yaml"]) + monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(catalog_path)) + result = CliRunner().invoke( + app, + [ + "route", + "scaffold", + "--input", + str(source), + "--output-dir", + str(tmp_path / "stage"), + ], + ) + assert result.exit_code == 0, result.output + assert ( + load_catalog(tmp_path / "stage/catalog.yaml").entries[0].id + == draft_request["lane_id"] + ) + source.write_text("title: [ghp_private\n") + result = CliRunner().invoke( + app, + [ + "route", + "scaffold", + "--input", + str(source), + "--output-dir", + str(tmp_path / "bad"), + ], + ) + assert result.exit_code == 1 + assert "ghp_private" not in result.output diff --git a/wiki/CatalogLaneScaffoldContract.md b/wiki/CatalogLaneScaffoldContract.md index f8faaf2..da6430a 100644 --- a/wiki/CatalogLaneScaffoldContract.md +++ b/wiki/CatalogLaneScaffoldContract.md @@ -80,3 +80,16 @@ platform credential-change validator and CCR-2026-0031 (metadata only). Knowledge: emergent-security 0.2.6, start and local-work, verified-files, orientation-only; package digest `8d1d6a9d4d07fec7b619618de7dfe2be2254c1790f85f4198c7fe19db9751c02`. + + +## Offline catalog command — 2026-10-04 + +`warden route scaffold --input --output-dir ` +validates the closed v1 request against catalog entry rules and refuses an +existing catalog lane ID or staging destination. The request fields above also +require explicit `reviewed: YYYY-MM-DD` text for deterministic review metadata. +Explicit `ungraded` risk is retained with the catalog's existing fail-safe meaning; +no default risk or workload identity is inferred. Outputs are `catalog.yaml`, +`playbook.md` and the unchanged semantic request in `request.yaml`. The output +catalog is draft with execution disabled and no fetch command. This command +does not invoke a platform renderer or replace the owner's admission lifecycle.