diff --git a/.gitignore b/.gitignore index 97f5134..ddae26a 100644 --- a/.gitignore +++ b/.gitignore @@ -175,3 +175,4 @@ cython_debug/ .pypirc *.swp +.claude/ralph-loop.local.md diff --git a/README.md b/README.md index 965f20d..daa5144 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,21 @@ Production uses the `vault` backend against OpenBao or HashiCorp Vault (Vault-co SSH secrets engine API). Template: `examples/warden.production.example.yaml`. See `wiki/OpsWardenConfig.md` and `wiki/OpenBaoSshEngineChecklist.md`. +## Routing lookup (`warden route`) + +ops-warden issues SSH certs and **routes** every other credential need to its +owner. The `route` command group is a read-only lookup over the pointer catalog +(`registry/routing/catalog.yaml`) — it never calls another subsystem or returns +secrets. + +```bash +warden route list [--all] [--json] # scenarios (active-only unless --all) +warden route show [--json] # owner + wiki/canon pointers; SSH adds steps +warden route find "issue an api key" # rank scenarios by keyword overlap +``` + +Full role and examples: `wiki/AccessRouting.md`. + ## Development ```bash diff --git a/SCOPE.md b/SCOPE.md index b363a3c..1b55349 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -21,8 +21,9 @@ ops-warden **issues short-lived SSH certificates and routes every other credenti need to the subsystem that owns it.** SSH signing is **production-verified** on Railiance OpenBao (`warden sign` against `https://bao.coulomb.social`, host CA trust deployed). The routing material — `wiki/AccessRouting.md`, the credential routing -wiki, NetKingdom security map, and a machine-readable pointer catalog -(`registry/routing/catalog.yaml`, WARDEN-WP-0010) — is operational. The opt-in +wiki, NetKingdom security map, a machine-readable pointer catalog +(`registry/routing/catalog.yaml`, WARDEN-WP-0010), and the `warden route` +lookup CLI over it (`list`/`show`/`find`, WARDEN-WP-0011) — is operational. The opt-in flex-auth pre-sign gate is **coded but off in production** until flex-auth publishes `ssh-certificate` policies (WARDEN-WP-0009). @@ -60,12 +61,12 @@ Full gap analysis: `history/2026-06-18-post-wp0008-intent-scope-reassessment.md` | NetKingdom evolution reflected in docs | Met | | Non-SSH secrets stay out of ops-warden | Met | -**Maturity vector:** `D5 / A3 / C4 / R3` (Discovery / Availability / Completeness / Reliability) +**Maturity vector:** `D5 / A4 / C4 / R3` (Discovery / Availability / Completeness / Reliability) | Dimension | Level | Meaning today | | --- | --- | --- | | D5 | Discovery | Routing wiki + security map + pointer catalog + NK canon cross-links | -| A3 | Availability | CLI + opt-in policy gate + machine-readable routing catalog; `warden route` lookup (A4) lands with WARDEN-WP-0011 | +| A4 | Availability | CLI + opt-in policy gate + `warden route` lookup over the machine-readable catalog (`list`/`show`/`find`, `--json` for agents) | | C4 | Completeness | SSH lane prod-verified; flex-auth policies external | | R3 | Reliability | Live OpenBao sign evidence on Railiance | @@ -95,6 +96,7 @@ for the rest. - `cert_command`: `warden sign --pubkey ` → cert on stdout - TTL enforcement per `ActorType` (`adm` 48 h, `agt` 24 h, `atm` 8 h) - `warden status`, cleanup, scorecard, signatures log +- `warden route` lookup CLI (`list`/`show`/`find`, `--json`) over the pointer catalog - `warden issue` and `ops-ssh-wrapper` (local backend; vault uses sign-only) - Runbooks for OpenBao config and Inter-Hub bootstrap SSH envelope @@ -113,13 +115,13 @@ for the rest. | WP-0007 | Opt-in flex-auth policy gate (`policy.enabled`) | | WP-0008 | Production sign verification, stewardship closeout, archive hygiene | | WP-0010 | "Issue SSH, route the rest" wording + `wiki/AccessRouting.md` + pointer catalog | +| WP-0011 | `warden route` lookup CLI (`list`/`show`/`find`) over the pointer catalog (A3 → A4) | ### Active / wait | WP | Status | Focus | | --- | --- | --- | -| **WP-0009** | `wait` | flex-auth `ssh-certificate` policies + `policy.enabled` production smoke | -| **WP-0011** | `ready` | `warden route` lookup CLI over the pointer catalog (A3 → A4) | +| **WP-0009** | `blocked` | flex-auth `ssh-certificate` policies + `policy.enabled` production smoke | | **WP-0012** | `backlog` | Routing scenario playbooks (draft until owner paths ship) | ### Known gaps (not yet workplanned) diff --git a/src/warden/cli.py b/src/warden/cli.py index c7698c2..492fa7d 100644 --- a/src/warden/cli.py +++ b/src/warden/cli.py @@ -23,6 +23,11 @@ app = typer.Typer( ) inventory_app = typer.Typer(help="Manage principals inventory", no_args_is_help=True) app.add_typer(inventory_app, name="inventory") +route_app = typer.Typer( + help="Look up which subsystem owns a credential need (read-only pointer layer)", + no_args_is_help=True, +) +app.add_typer(route_app, name="route") console = Console() err = Console(stderr=True) @@ -512,3 +517,143 @@ def log( e.get("backend", ""), ) console.print(table) + + +# --------------------------------------------------------------------------- +# warden route — read-only routing lookup over the pointer catalog +# --------------------------------------------------------------------------- + +def _load_catalog(): + from warden.routing import CatalogError, load_catalog + try: + return load_catalog() + except CatalogError as e: + err.print(f"[red]Routing catalog error:[/red] {e}") + raise typer.Exit(1) + + +def _entry_summary(entry) -> dict: + """Pointer-only summary. Never includes secret material.""" + return { + "id": entry.id, + "title": entry.title, + "owner_repo": entry.owner_repo, + "subsystem": entry.subsystem, + "warden_executes": entry.warden_executes, + "wiki_ref": entry.wiki_ref, + "canon_ref": entry.canon_ref, + "reviewed": entry.reviewed, + "status": entry.status, + } + + +def _print_entry_table(entries, title: str) -> None: + table = Table(title=title) + table.add_column("ID") + table.add_column("Need") + table.add_column("Owner") + table.add_column("warden") + table.add_column("Status") + for e in entries: + executes = "[green]issue[/green]" if e.warden_executes else "route" + status_styled = e.status if e.status == "active" else f"[yellow]{e.status}[/yellow]" + table.add_row(e.id, e.title, e.owner_repo, executes, status_styled) + console.print(table) + + +@route_app.command("list") +def route_list( + output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False, + all_entries: Annotated[bool, typer.Option("--all", help="Include draft entries")] = False, + tag: Annotated[Optional[str], typer.Option("--tag", help="Filter by need keyword")] = None, +) -> None: + """List routing scenarios. Active-only unless --all.""" + catalog = _load_catalog() + entries = catalog.listed(include_draft=all_entries) + if tag: + t = tag.lower() + entries = [e for e in entries if t in [k.lower() for k in e.need_keywords]] + + if output_json: + print(json.dumps([_entry_summary(e) for e in entries], indent=2)) + return + + if not entries: + console.print("No matching routing entries.") + return + _print_entry_table(entries, "Routing scenarios") + + +@route_app.command("show") +def route_show( + entry_id: Annotated[str, typer.Argument(help="Catalog entry id (see `warden route list`)")], + output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False, +) -> None: + """Show owner, pointers, and (SSH only) the authored steps for one scenario.""" + catalog = _load_catalog() + entry = catalog.get(entry_id) + if entry is None: + err.print( + f"[red]Unknown routing id {entry_id!r}.[/red] " + f"Try: warden route find {entry_id!r}" + ) + raise typer.Exit(1) + + if output_json: + summary = _entry_summary(entry) + summary["need_keywords"] = entry.need_keywords + if entry.warden_executes: + summary["steps"] = entry.steps + summary["cert_command"] = entry.cert_command + else: + summary["next_action"] = ( + f"next action on `{entry.owner_repo}` — see `{entry.wiki_ref}`" + ) + print(json.dumps(summary, indent=2)) + return + + console.print(f"[bold]{entry.title}[/bold] ([cyan]{entry.id}[/cyan])") + console.print(f" owner : {entry.owner_repo} ({entry.subsystem})") + console.print(f" wiki : {entry.wiki_ref}") + console.print(f" canon : {entry.canon_ref}") + console.print(f" reviewed : {entry.reviewed} status: {entry.status}") + + if entry.warden_executes: + console.print("\n[green]ops-warden issues this directly.[/green]") + console.print(f" cert_command: [bold]{entry.cert_command}[/bold]") + if entry.steps: + console.print(" steps:") + for i, step in enumerate(entry.steps, 1): + console.print(f" {i}. {step}") + console.print( + " precondition: actor in inventory? backend configured? run `warden status`." + ) + else: + console.print( + f"\n[yellow]ops-warden does not issue this.[/yellow] " + f"Next action on [bold]{entry.owner_repo}[/bold] — see {entry.wiki_ref}." + ) + + +@route_app.command("find") +def route_find( + query: Annotated[str, typer.Argument(help="Free-text need, e.g. 'issue core api key'")], + output_json: Annotated[bool, typer.Option("--json", help="Output JSON")] = False, + all_entries: Annotated[bool, typer.Option("--all", help="Include draft entries")] = False, + limit: Annotated[int, typer.Option("--limit", help="Max matches")] = 5, +) -> None: + """Rank routing scenarios by keyword overlap with the query.""" + catalog = _load_catalog() + matches = catalog.find(query, include_draft=all_entries, limit=limit) + + if output_json: + print(json.dumps([_entry_summary(e) for e in matches], indent=2)) + return + + if not matches: + console.print( + f"No routing match for {query!r}. " + "Try `warden route list --all` to browse all scenarios." + ) + return + _print_entry_table(matches, f"Matches for {query!r}") diff --git a/src/warden/routing/__init__.py b/src/warden/routing/__init__.py new file mode 100644 index 0000000..3a3bf53 --- /dev/null +++ b/src/warden/routing/__init__.py @@ -0,0 +1,17 @@ +"""Routing lookup — read-only pointer layer over registry/routing/catalog.yaml. + +This package never calls OpenBao, flex-auth, key-cape, ops-bridge, or any other +subsystem. It loads the machine-readable routing catalog and answers "who owns +this need and where is the authoritative doc". The one lane ops-warden executes +(SSH certificate issuance) is the only entry that carries authored steps. +""" +from warden.routing.catalog import Catalog, CatalogError, find_catalog_path, load_catalog +from warden.routing.models import RouteEntry + +__all__ = [ + "Catalog", + "CatalogError", + "RouteEntry", + "find_catalog_path", + "load_catalog", +] diff --git a/src/warden/routing/catalog.py b/src/warden/routing/catalog.py new file mode 100644 index 0000000..7111bd7 --- /dev/null +++ b/src/warden/routing/catalog.py @@ -0,0 +1,171 @@ +"""Load and validate the routing pointer catalog. + +The catalog lives at ``registry/routing/catalog.yaml`` in the repo root. Resolution +order: + +1. ``WARDEN_ROUTING_CATALOG`` env var, if set (used by tests / overrides). +2. Walk upward from this module looking for ``registry/routing/catalog.yaml``. + +Validation enforces the **no-double-source rule**: only ``warden_executes: true`` +entries may carry an authored ``steps`` block or a ``cert_command``. Any non-SSH +entry that does so is a validation error — ops-warden points at the owner's doc, it +never restates another subsystem's procedure. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import List, Optional + +import yaml + +from warden.routing.models import RouteEntry + +_REQUIRED_FIELDS = ( + "id", + "title", + "need_keywords", + "owner_repo", + "subsystem", + "warden_executes", + "wiki_ref", + "canon_ref", + "reviewed", + "status", +) +_VALID_STATUS = ("active", "draft") + + +class CatalogError(Exception): + """Raised when the routing catalog is missing or invalid.""" + + +def find_catalog_path(start: Optional[Path] = None) -> Path: + """Locate registry/routing/catalog.yaml. + + Honors WARDEN_ROUTING_CATALOG first; otherwise walks up from `start` + (default: this module) until a repo root containing the catalog is found. + """ + override = os.environ.get("WARDEN_ROUTING_CATALOG") + if override: + return Path(os.path.expanduser(override)) + + rel = Path("registry") / "routing" / "catalog.yaml" + here = (start or Path(__file__)).resolve() + for parent in [here, *here.parents]: + candidate = parent / rel + if candidate.exists(): + return candidate + raise CatalogError( + f"Routing catalog not found ({rel}). Set WARDEN_ROUTING_CATALOG to override." + ) + + +@dataclass +class Catalog: + path: Path + entries: List[RouteEntry] + + # --- lookup helpers --------------------------------------------------- + + def get(self, entry_id: str) -> Optional[RouteEntry]: + for e in self.entries: + if e.id == entry_id: + return e + return None + + def listed(self, include_draft: bool = False) -> List[RouteEntry]: + if include_draft: + return list(self.entries) + return [e for e in self.entries if e.is_active] + + def find(self, query: str, include_draft: bool = False, limit: int = 5) -> List[RouteEntry]: + """Rank entries by keyword overlap with the query. Highest first.""" + tokens = [t for t in query.lower().replace("-", " ").split() if t] + pool = self.listed(include_draft=include_draft) + scored = [(e.match_score(tokens), e) for e in pool] + scored = [(s, e) for s, e in scored if s > 0] + scored.sort(key=lambda pair: (-pair[0], pair[1].id)) + return [e for _, e in scored[:limit]] + + +def _parse_entry(raw: dict, index: int) -> RouteEntry: + if not isinstance(raw, dict): + raise CatalogError(f"entry #{index} is not a mapping") + + missing = [f for f in _REQUIRED_FIELDS if f not in raw] + if missing: + ident = raw.get("id", f"#{index}") + raise CatalogError(f"entry {ident!r} missing required field(s): {', '.join(missing)}") + + warden_executes = bool(raw["warden_executes"]) + steps = raw.get("steps") or [] + cert_command = raw.get("cert_command") + status = str(raw["status"]) + + if status not in _VALID_STATUS: + raise CatalogError( + f"entry {raw['id']!r} has invalid status {status!r} (expected one of {_VALID_STATUS})" + ) + + # No-double-source rule: authored procedure only on the SSH lane. + if not warden_executes and steps: + raise CatalogError( + f"entry {raw['id']!r} is not warden_executes but carries a `steps` block " + "— routed needs point at the owner's doc; they must not restate procedure " + "(no-double-source rule)." + ) + if not warden_executes and cert_command: + raise CatalogError( + f"entry {raw['id']!r} is not warden_executes but carries a `cert_command`." + ) + + if not isinstance(raw["need_keywords"], list): + raise CatalogError(f"entry {raw['id']!r} need_keywords must be a list") + + return RouteEntry( + id=str(raw["id"]), + title=str(raw["title"]), + need_keywords=[str(k) for k in raw["need_keywords"]], + owner_repo=str(raw["owner_repo"]), + subsystem=str(raw["subsystem"]), + warden_executes=warden_executes, + wiki_ref=str(raw["wiki_ref"]), + canon_ref=str(raw["canon_ref"]), + reviewed=str(raw["reviewed"]), + status=status, + steps=[str(s) for s in steps], + cert_command=str(cert_command) if cert_command else None, + ) + + +def load_catalog(path: Optional[Path] = None) -> Catalog: + """Load, parse, and validate the routing catalog.""" + catalog_path = path or find_catalog_path() + if not catalog_path.exists(): + raise CatalogError(f"Routing catalog not found: {catalog_path}") + + try: + with catalog_path.open() as f: + raw = yaml.safe_load(f) + except yaml.YAMLError as e: + raise CatalogError(f"Invalid YAML in {catalog_path}: {e}") from e + + if not isinstance(raw, dict): + raise CatalogError("Catalog must be a YAML mapping") + + raw_entries = raw.get("entries") + if not isinstance(raw_entries, list) or not raw_entries: + raise CatalogError("Catalog has no `entries` list") + + entries: List[RouteEntry] = [] + seen: set[str] = set() + for i, raw_entry in enumerate(raw_entries): + entry = _parse_entry(raw_entry, i) + if entry.id in seen: + raise CatalogError(f"duplicate entry id: {entry.id!r}") + seen.add(entry.id) + entries.append(entry) + + return Catalog(path=catalog_path, entries=entries) diff --git a/src/warden/routing/models.py b/src/warden/routing/models.py new file mode 100644 index 0000000..868b2a1 --- /dev/null +++ b/src/warden/routing/models.py @@ -0,0 +1,49 @@ +"""Data model for routing catalog entries. + +A `RouteEntry` is a pointer: it names the owner and the authoritative doc for a +credential need. Only the SSH lane (`warden_executes: true`) may carry an authored +`steps` block and a `cert_command` pattern — every other entry is identifiers and +pointers only (the no-double-source rule, enforced in `catalog.py`). +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import List, Optional + + +@dataclass +class RouteEntry: + id: str + title: str + need_keywords: List[str] + owner_repo: str + subsystem: str + warden_executes: bool + wiki_ref: str + canon_ref: str + reviewed: str + status: str # "active" | "draft" + # SSH lane only — None/empty for routed (non-executed) needs. + steps: List[str] = field(default_factory=list) + cert_command: Optional[str] = None + + @property + def is_active(self) -> bool: + return self.status == "active" + + def match_score(self, tokens: List[str]) -> int: + """Keyword-overlap score against need_keywords, title, and id. + + Pure ranking helper — no I/O, no external calls. + """ + haystack = set(k.lower() for k in self.need_keywords) + haystack.update(self.id.lower().replace("-", " ").split()) + haystack.update(self.title.lower().replace("-", " ").split()) + score = 0 + for tok in tokens: + t = tok.lower() + if t in haystack: + score += 2 + elif any(t in h or h in t for h in haystack): + score += 1 + return score diff --git a/tests/test_routing.py b/tests/test_routing.py new file mode 100644 index 0000000..1ee596c --- /dev/null +++ b/tests/test_routing.py @@ -0,0 +1,225 @@ +"""Tests for the routing pointer catalog and `warden route` CLI. + +No test here requires a live subsystem — routing is a read-only pointer layer. +""" +import json +import re +from pathlib import Path + +import pytest +import yaml +from typer.testing import CliRunner + +from warden.cli import app +from warden.routing import CatalogError, load_catalog +from warden.routing.catalog import find_catalog_path + +runner = CliRunner() + + +def _repo_catalog() -> Path: + return find_catalog_path() + + +def _write_catalog(tmp_path: Path, entries: list[dict]) -> Path: + path = tmp_path / "catalog.yaml" + path.write_text(yaml.dump({"version": 1, "entries": entries})) + return path + + +SSH_ENTRY = { + "id": "ssh-cert-host-access", + "title": "SSH cert", + "need_keywords": ["ssh", "cert", "sign"], + "owner_repo": "ops-warden", + "subsystem": "ops-warden", + "warden_executes": True, + "wiki_ref": "wiki/AccessRouting.md#issue-vs-route", + "canon_ref": "net-kingdom/docs/x.md", + "reviewed": "2026-06-18", + "status": "active", + "cert_command": "warden sign --pubkey ", + "steps": ["confirm inventory", "sign"], +} + +ROUTED_ENTRY = { + "id": "openbao-api-key", + "title": "API key", + "need_keywords": ["api", "key", "openbao"], + "owner_repo": "railiance-platform", + "subsystem": "OpenBao", + "warden_executes": False, + "wiki_ref": "wiki/CredentialRouting.md#routing-table", + "canon_ref": "net-kingdom/docs/x.md", + "reviewed": "2026-06-18", + "status": "active", +} + + +# --------------------------------------------------------------------------- +# Catalog load + validation +# --------------------------------------------------------------------------- + +def test_real_catalog_loads(): + catalog = load_catalog(_repo_catalog()) + assert len(catalog.entries) >= 6 + ssh = catalog.get("ssh-cert-host-access") + assert ssh is not None and ssh.warden_executes is True + assert ssh.cert_command and "warden sign" in ssh.cert_command + + +def test_real_catalog_has_one_executed_lane(): + catalog = load_catalog(_repo_catalog()) + executed = [e for e in catalog.entries if e.warden_executes] + assert [e.id for e in executed] == ["ssh-cert-host-access"] + + +def test_no_double_source_rule_rejects_routed_steps(tmp_path): + bad = dict(ROUTED_ENTRY) + bad["steps"] = ["do a thing on OpenBao"] # non-SSH entry must not carry steps + path = _write_catalog(tmp_path, [SSH_ENTRY, bad]) + with pytest.raises(CatalogError, match="no-double-source"): + load_catalog(path) + + +def test_routed_cert_command_rejected(tmp_path): + bad = dict(ROUTED_ENTRY) + bad["cert_command"] = "warden secret get" + path = _write_catalog(tmp_path, [bad]) + with pytest.raises(CatalogError, match="cert_command"): + load_catalog(path) + + +def test_duplicate_id_rejected(tmp_path): + path = _write_catalog(tmp_path, [ROUTED_ENTRY, dict(ROUTED_ENTRY)]) + with pytest.raises(CatalogError, match="duplicate"): + load_catalog(path) + + +def test_missing_field_rejected(tmp_path): + bad = {k: v for k, v in ROUTED_ENTRY.items() if k != "owner_repo"} + path = _write_catalog(tmp_path, [bad]) + with pytest.raises(CatalogError, match="owner_repo"): + load_catalog(path) + + +def test_missing_catalog_file(): + with pytest.raises(CatalogError): + load_catalog(Path("/nonexistent/catalog.yaml")) + + +# --------------------------------------------------------------------------- +# find ranking +# --------------------------------------------------------------------------- + +def test_find_active_excludes_draft(): + catalog = load_catalog(_repo_catalog()) + ids = [e.id for e in catalog.find("issue core api key")] + assert "issue-core-ingestion-api-key" not in ids + + +def test_find_all_includes_draft(): + catalog = load_catalog(_repo_catalog()) + ids = [e.id for e in catalog.find("issue core api key", include_draft=True)] + assert "issue-core-ingestion-api-key" in ids + + +def test_find_ssh_tunnel_top_match(): + catalog = load_catalog(_repo_catalog()) + matches = catalog.find("ssh tunnel") + assert matches and matches[0].id == "ops-bridge-tunnel" + + +# --------------------------------------------------------------------------- +# CLI (uses the repo catalog via env override) +# --------------------------------------------------------------------------- + +@pytest.fixture +def repo_catalog_env(monkeypatch): + monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog())) + + +def test_cli_list_active_only(repo_catalog_env): + result = runner.invoke(app, ["route", "list", "--json"]) + assert result.exit_code == 0 + ids = [e["id"] for e in json.loads(result.stdout)] + assert "issue-core-ingestion-api-key" not in ids + + +def test_cli_list_all_includes_draft(repo_catalog_env): + result = runner.invoke(app, ["route", "list", "--all", "--json"]) + ids = [e["id"] for e in json.loads(result.stdout)] + assert "issue-core-ingestion-api-key" in ids + + +def test_cli_show_ssh_json_includes_cert_pattern(repo_catalog_env): + result = runner.invoke(app, ["route", "show", "ssh-cert-host-access", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.stdout) + assert data["warden_executes"] is True + assert "warden sign" in data["cert_command"] + assert data["steps"] + + +def test_cli_show_routed_has_next_action_not_steps(repo_catalog_env): + result = runner.invoke(app, ["route", "show", "openbao-api-key", "--json"]) + data = json.loads(result.stdout) + assert data["warden_executes"] is False + assert "steps" not in data + assert "next_action" in data + + +def test_cli_show_unknown_exits_one(repo_catalog_env): + result = runner.invoke(app, ["route", "show", "does-not-exist"]) + assert result.exit_code == 1 + + +def test_cli_find_json(repo_catalog_env): + result = runner.invoke(app, ["route", "find", "ssh tunnel", "--json"]) + assert result.exit_code == 0 + ids = [e["id"] for e in json.loads(result.stdout)] + assert "ops-bridge-tunnel" in ids + + +# --------------------------------------------------------------------------- +# T5 drift guard — every wiki_ref anchor resolves, every entry has a reviewed date +# --------------------------------------------------------------------------- + +def _github_slug(heading: str) -> str: + """Approximate GitHub's heading-anchor slug algorithm.""" + text = heading.strip().lower() + text = re.sub(r"[^\w\s-]", "", text) # drop punctuation (em-dash, parens, etc.) + text = text.replace(" ", "-") + return text + + +def _heading_anchors(md_path: Path) -> set[str]: + anchors: set[str] = set() + for line in md_path.read_text().splitlines(): + m = re.match(r"^#{1,6}\s+(.*)$", line) + if m: + anchors.add(_github_slug(m.group(1))) + return anchors + + +def test_every_wiki_ref_anchor_resolves(): + catalog = load_catalog(_repo_catalog()) + repo_root = _repo_catalog().parents[2] # registry/routing/catalog.yaml -> repo root + failures = [] + for entry in catalog.entries: + rel, _, anchor = entry.wiki_ref.partition("#") + md_path = repo_root / rel + if not md_path.exists(): + failures.append(f"{entry.id}: wiki file missing: {rel}") + continue + if anchor and anchor not in _heading_anchors(md_path): + failures.append(f"{entry.id}: anchor #{anchor} not found in {rel}") + assert not failures, "\n".join(failures) + + +def test_every_entry_has_reviewed_date(): + catalog = load_catalog(_repo_catalog()) + for entry in catalog.entries: + assert re.match(r"^\d{4}-\d{2}-\d{2}$", entry.reviewed), ( + f"{entry.id}: reviewed must be YYYY-MM-DD, got {entry.reviewed!r}" + ) diff --git a/wiki/AccessRouting.md b/wiki/AccessRouting.md index 83b37b8..4463819 100644 --- a/wiki/AccessRouting.md +++ b/wiki/AccessRouting.md @@ -57,15 +57,48 @@ the owner's runbook. See the no-double-source rule in --- +## Routing lookup CLI (`warden route`) + +Agents and operators query the pointer catalog directly instead of re-deriving +routing from wiki prose. The command group is **read-only** — it never calls +OpenBao, flex-auth, key-cape, or any other subsystem, and never returns secret +material. + +```bash +warden route list [--json] [--all] [--tag ] # active-only unless --all +warden route show [--json] # owner + pointers; SSH adds steps +warden route find "" [--json] [--all] # rank by keyword overlap +``` + +Agent-oriented examples: + +```bash +# "I need an API key" — find the owner, get a pointer, act there yourself +warden route find "openrouter api key" --json +warden route show openbao-api-key --json +# → {"warden_executes": false, "next_action": "next action on `railiance-platform` — see `wiki/CredentialRouting.md#routing-table`"} + +# The one lane ops-warden executes: SSH. `show` appends the authored steps + cert pattern. +warden route show ssh-cert-host-access --json +# → {"warden_executes": true, "cert_command": "warden sign --pubkey ", "steps": [...]} +``` + +`show` on a routed (non-SSH) need always ends with **"next action on +`` — see ``"** and never implies ops-warden performed +anything. Draft scenarios (owner path not yet shipped) are hidden unless `--all`. + +--- + ## Audience notes - **Human operators** read this page and `CredentialRouting.md` to choose the right subsystem, then follow that subsystem's own docs. -- **Agents / CI** will read the machine-readable routing catalog - (`registry/routing/catalog.yaml`, surfaced via `warden route` — WARDEN-WP-0011) - so routing does not have to be re-derived from wiki prose each session. +- **Agents / CI** read the machine-readable routing catalog + (`registry/routing/catalog.yaml`) via `warden route` (above) so routing does + not have to be re-derived from wiki prose each session. - **Same truth, two shapes:** humans read the wiki; agents read the catalog. The - catalog references wiki sections by anchor so the two cannot drift apart. + catalog references wiki sections by anchor so the two cannot drift apart — a + test (`tests/test_routing.py`) fails CI if any `wiki_ref` anchor stops resolving. --- diff --git a/workplans/WARDEN-WP-0011-routing-guide-cli.md b/workplans/WARDEN-WP-0011-routing-guide-cli.md index 8fb57ff..b31ef38 100644 --- a/workplans/WARDEN-WP-0011-routing-guide-cli.md +++ b/workplans/WARDEN-WP-0011-routing-guide-cli.md @@ -4,7 +4,7 @@ type: workplan title: "Routing Lookup CLI" domain: custodian repo: ops-warden -status: ready +status: done owner: codex topic_slug: custodian planning_priority: high @@ -69,72 +69,72 @@ foreign subsystems. SSH precondition hints live inside `show` instead. ```task id: WARDEN-WP-0011-T01 -status: todo +status: done priority: high state_hub_task_id: "55b8422c-ad3c-4084-9e00-acaa4c360906" ``` -- [ ] Add `src/warden/routing/` package: `models.py`, `catalog.py`. -- [ ] Load and validate `registry/routing/catalog.yaml`. -- [ ] Enforce the no-double-source rule: non-SSH entries with a `steps` block are a +- [x] Add `src/warden/routing/` package: `models.py`, `catalog.py`. +- [x] Load and validate `registry/routing/catalog.yaml`. +- [x] Enforce the no-double-source rule: non-SSH entries with a `steps` block are a validation error. Clear errors for missing file, schema violations, dup `id`. ### T2 — `warden route list` and `show` ```task id: WARDEN-WP-0011-T02 -status: todo +status: done priority: high state_hub_task_id: "60b679c5-79bd-4186-b5a6-ac576931f06c" ``` -- [ ] Register `route` Typer sub-app on the main CLI. -- [ ] `list` — Rich table + `--json` array of summaries; active-only unless `--all`. -- [ ] `show` — owner, prerequisites, pointers (`wiki_ref`, `canon_ref`), +- [x] Register `route` Typer sub-app on the main CLI. +- [x] `list` — Rich table + `--json` array of summaries; active-only unless `--all`. +- [x] `show` — owner, prerequisites, pointers (`wiki_ref`, `canon_ref`), `warden_executes`, anti-patterns; SSH entries also append `steps` + cert pattern. -- [ ] Exit 1 with a `find` hint when `show` id is unknown. +- [x] Exit 1 with a `find` hint when `show` id is unknown. ### T3 — `warden route find` ```task id: WARDEN-WP-0011-T03 -status: todo +status: done priority: high state_hub_task_id: "d307701f-0117-44f0-80fd-ca6f7ae06f42" ``` -- [ ] Tokenize query; match against `need_keywords`, `title`, `id`. -- [ ] Rank, show top matches (default 5); `--json` for agents. -- [ ] Fixtures: "issue core api key", "ssh tunnel", "openrouter key". +- [x] Tokenize query; match against `need_keywords`, `title`, `id`. +- [x] Rank, show top matches (default 5); `--json` for agents. +- [x] Fixtures: "issue core api key", "ssh tunnel", "openrouter key". ### T4 — Tests ```task id: WARDEN-WP-0011-T04 -status: todo +status: done priority: high state_hub_task_id: "00a76e0f-8ab6-4f9a-ac6a-00eae633342c" ``` -- [ ] `tests/test_routing.py` — catalog load, no-double-source validation rejects a +- [x] `tests/test_routing.py` — catalog load, no-double-source validation rejects a non-SSH `steps` block, find ranking, show JSON shape, SSH `show` includes cert pattern. -- [ ] No integration test requires a live subsystem. +- [x] No integration test requires a live subsystem. ### T5 — Doc consistency + drift guard ```task id: WARDEN-WP-0011-T05 -status: todo +status: done priority: high state_hub_task_id: "bf848375-eca7-4116-bb1d-fb7df6395c70" ``` -- [ ] CI/test: every `wiki_ref` anchor resolves to an existing in-repo wiki section; +- [x] CI/test: every `wiki_ref` anchor resolves to an existing in-repo wiki section; every entry has a `reviewed` date. -- [ ] `wiki/AccessRouting.md` — CLI section with agent-oriented examples. -- [ ] README — `warden route --help` quick reference. -- [ ] Bump SCOPE availability note A3 → A4 on ship. +- [x] `wiki/AccessRouting.md` — CLI section with agent-oriented examples. +- [x] README — `warden route --help` quick reference. +- [x] Bump SCOPE availability note A3 → A4 on ship. ---