tegwick
4fee839b11
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663
1014 lines
38 KiB
Python
1014 lines
38 KiB
Python
"""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 datetime import date
|
|
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 days_since_review, find_catalog_path, is_review_stale
|
|
from warden.scorecard import check_catalog_rotation_coverage
|
|
|
|
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",
|
|
"workload_ref": {
|
|
"applicability": "not-applicable",
|
|
"reason": "generic certificate action",
|
|
},
|
|
"cert_command": "warden sign <actor> --pubkey <path>",
|
|
"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",
|
|
"workload_ref": {
|
|
"applicability": "not-applicable",
|
|
"reason": "generic credential pattern",
|
|
},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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_every_catalog_lane_declares_workload_applicability():
|
|
catalog = load_catalog(_repo_catalog())
|
|
assert all(entry.workload_ref is not None for entry in catalog.entries)
|
|
assert {entry.workload_ref.resolution for entry in catalog.entries} == {
|
|
"resolved", "unknown", "not-applicable"
|
|
}
|
|
|
|
|
|
def test_managed_and_operational_workload_references_parse():
|
|
catalog = load_catalog(_repo_catalog())
|
|
managed = catalog.get("issue-core-ingestion-api-key").workload_ref
|
|
assert managed.resolution == "resolved"
|
|
assert (managed.rapp_id, managed.name, managed.deployable) == (
|
|
"rapp-issue-core", "issue-core", "issue-core"
|
|
)
|
|
operational = catalog.get("ops-warden-warden-sign-token").workload_ref
|
|
assert operational.resolution == "resolved"
|
|
assert operational.rapp_id is None
|
|
assert operational.name == "ops-warden"
|
|
assert operational.declaration_ref == "tenancy.yaml"
|
|
|
|
|
|
def test_workload_reference_rejects_ambiguous_absence(tmp_path):
|
|
bad = dict(ROUTED_ENTRY)
|
|
bad.pop("workload_ref")
|
|
with pytest.raises(CatalogError, match="workload_ref"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_workload_reference_rejects_malformed_managed_target(tmp_path):
|
|
bad = dict(ROUTED_ENTRY)
|
|
bad["workload_ref"] = {
|
|
"applicability": "applicable",
|
|
"rapp_id": "rapp-issue-core",
|
|
}
|
|
with pytest.raises(CatalogError, match="requires name"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_ops_warden_warden_sign_lane_has_native_exec():
|
|
"""RAILIANCE-WP-0005 T08 — broker lane routes to railiance-platform credential exec."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
e = catalog.get("ops-warden-warden-sign-token")
|
|
assert e is not None and e.is_active and e.owner_repo == "railiance-platform"
|
|
assert e.has_native_exec is True
|
|
assert e.exec_owner == "railiance-platform"
|
|
assert "credential.py exec" in e.exec_command
|
|
assert "ops-warden/warden-sign" in e.exec_command
|
|
assert "credential-exec-ops-warden-smoke" in e.pointer_command
|
|
assert e.warden_executes is False
|
|
assert e.resolvable is False # broker lane — owner exec, not warden access --fetch
|
|
|
|
|
|
def test_route_find_vault_token_ops_warden_prefers_broker_lane():
|
|
catalog = load_catalog(_repo_catalog())
|
|
matches = catalog.find("VAULT_TOKEN ops-warden warden sign", limit=3)
|
|
assert matches[0].id == "ops-warden-warden-sign-token"
|
|
|
|
|
|
def test_whynot_design_npm_lane_is_concrete_and_resolvable():
|
|
"""The provisioned npm publish lane has no placeholders and reports resolvable."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
e = catalog.get("whynot-design-npm-publish")
|
|
assert e is not None and e.is_active and e.exec_capable
|
|
assert e.resolvable is True
|
|
assert "<" not in e.fetch_command and ">" not in e.fetch_command
|
|
assert "platform/workloads/coulomb/whynot-design/npm-publish" in e.fetch_command
|
|
|
|
|
|
def test_policy_nexus_source_read_lane_is_exact_high_risk_and_resolvable():
|
|
catalog = load_catalog(_repo_catalog())
|
|
entry = catalog.get("policy-nexus-forgejo-source-read")
|
|
assert entry is not None and entry.is_active and entry.exec_capable
|
|
assert entry.resolvable is True
|
|
assert entry.risk == "high"
|
|
assert entry.owner_repo == "railiance-platform"
|
|
assert entry.fetch_command == (
|
|
"bao kv get -field=FORGEJO_SOURCE_TOKEN "
|
|
"platform/workloads/policy-nexus/forgejo-source-read"
|
|
)
|
|
assert entry.path_template == "platform/workloads/policy-nexus/forgejo-source-read"
|
|
assert entry.auth_method.endswith(
|
|
"role=policy-nexus-forgejo-source-workload-kv-read"
|
|
)
|
|
assert entry.delegation is not None and entry.delegation.mode == "native"
|
|
|
|
|
|
def test_route_find_policy_nexus_source_read_prefers_concrete_lane():
|
|
catalog = load_catalog(_repo_catalog())
|
|
matches = catalog.find(
|
|
"policy nexus Forgejo private source repository read token Actions", limit=1
|
|
)
|
|
assert matches[0].id == "policy-nexus-forgejo-source-read"
|
|
|
|
|
|
def test_generic_and_template_lanes_not_resolvable():
|
|
catalog = load_catalog(_repo_catalog())
|
|
# generic openbao lane has <FIELD>/<path_template>; login lane has <domain>.
|
|
assert catalog.get("openbao-api-key").resolvable is False
|
|
assert catalog.get("key-cape-oidc-login").resolvable is False
|
|
|
|
|
|
def test_platform_admin_login_lane_is_exact_and_non_value_bearing():
|
|
entry = load_catalog(_repo_catalog()).get("openbao-platform-admin-login")
|
|
assert entry.lane == "login"
|
|
assert entry.risk == "high"
|
|
assert entry.fetch_command == (
|
|
"bao login -no-print -method=oidc -path=netkingdom role=platform-admin"
|
|
)
|
|
assert entry.workload_ref.resolution == "not-applicable"
|
|
|
|
|
|
def test_netkingdom_sso_bind_lanes_are_routed_but_not_resolvable():
|
|
catalog = load_catalog(_repo_catalog())
|
|
for lane_id in (
|
|
"net-kingdom-lldap-bind-credential",
|
|
"net-kingdom-privacyidea-admin-token",
|
|
):
|
|
entry = catalog.get(lane_id)
|
|
assert entry is not None
|
|
assert entry.owner_repo == "railiance-platform"
|
|
assert entry.risk == "high"
|
|
assert entry.warden_executes is False
|
|
assert entry.exec_capable is False
|
|
assert entry.resolvable is False
|
|
assert entry.delegation.blocked_on
|
|
assert "net-kingdom-sso-bind-credentials.md#worker-checklist" in entry.wiki_ref
|
|
|
|
|
|
def test_openbao_recovery_ceremony_is_non_value_bearing_owner_pointer():
|
|
entry = load_catalog(_repo_catalog()).get("openbao-shamir-recovery-ceremony")
|
|
assert entry.lane == "ceremony"
|
|
assert entry.risk == "high"
|
|
assert entry.owner_repo == "railiance-platform"
|
|
assert entry.warden_executes is False
|
|
assert entry.exec_capable is False
|
|
assert entry.has_handoff is False
|
|
assert entry.vends_secret is False
|
|
assert entry.workload_ref.resolution == "not-applicable"
|
|
|
|
|
|
def test_find_exact_id_wins_over_keyword_collision():
|
|
catalog = load_catalog(_repo_catalog())
|
|
# "npm" alone collides with openbao-api-key; the exact id must resolve uniquely.
|
|
assert catalog.find("whynot-design-npm-publish", limit=1)[0].id == "whynot-design-npm-publish"
|
|
|
|
|
|
def test_native_exec_owner_on_npm_lane():
|
|
"""secrets-engine is the owner-native exec front door for the npm lane (WP-0019)."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
e = catalog.get("whynot-design-npm-publish")
|
|
assert e.has_native_exec is True
|
|
assert e.exec_owner == "secrets-engine"
|
|
assert "secrets-engine exec --catalog whynot-design-npm-publish" in e.exec_command
|
|
assert "secrets-engine route" in e.pointer_command
|
|
# The proxy fallback is still available (exec_capable + resolvable).
|
|
assert e.exec_capable is True and e.resolvable is True
|
|
|
|
|
|
def test_lanes_without_native_exec():
|
|
catalog = load_catalog(_repo_catalog())
|
|
assert catalog.get("openbao-api-key").has_native_exec is False
|
|
assert catalog.get("ssh-cert-host-access").has_native_exec is False
|
|
|
|
|
|
def test_cli_show_native_exec_json(repo_catalog_env):
|
|
result = runner.invoke(app, ["route", "show", "whynot-design-npm-publish", "--json"])
|
|
data = json.loads(result.stdout)
|
|
assert data["exec_owner"] == "secrets-engine"
|
|
assert "secrets-engine exec" in data["exec_command"]
|
|
assert "primary" in data["next_action"] and "secrets-engine" in data["next_action"]
|
|
|
|
|
|
def test_cli_show_warden_sign_broker_json(repo_catalog_env):
|
|
result = runner.invoke(app, ["route", "show", "ops-warden-warden-sign-token", "--json"])
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.stdout)
|
|
assert data["owner_repo"] == "railiance-platform"
|
|
assert data["exec_owner"] == "railiance-platform"
|
|
assert "credential.py exec" in data["exec_command"]
|
|
assert "primary" in data["next_action"] and "railiance-platform" in data["next_action"]
|
|
|
|
|
|
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"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Structured handoff fields (WP-0014, T1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_handoff_fields_parse_on_routed_entry(tmp_path):
|
|
entry = dict(ROUTED_ENTRY)
|
|
entry["auth_method"] = "key-cape OIDC → bao login -method=oidc role=<domain>"
|
|
entry["path_template"] = "platform/workloads/<domain>/<workload>/<bundle>"
|
|
entry["fetch_command"] = "bao kv get -field=<FIELD> <path_template>"
|
|
entry["policy_ref"] = "flex-auth check secret.read:<domain>"
|
|
entry["exec_capable"] = True
|
|
catalog = load_catalog(_write_catalog(tmp_path, [entry]))
|
|
e = catalog.get("openbao-api-key")
|
|
assert e.has_handoff is True
|
|
assert e.exec_capable is True
|
|
assert e.path_template.startswith("platform/workloads/")
|
|
|
|
|
|
def test_real_catalog_openbao_entry_has_handoff():
|
|
e = load_catalog(_repo_catalog()).get("openbao-api-key")
|
|
assert e is not None and e.has_handoff and e.exec_capable
|
|
assert "<" in e.path_template and "<" in e.fetch_command # templates, not values
|
|
|
|
|
|
def test_exec_capable_without_fetch_command_rejected(tmp_path):
|
|
bad = dict(ROUTED_ENTRY)
|
|
bad["exec_capable"] = True # no fetch_command
|
|
with pytest.raises(CatalogError, match="fetch_command"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"leaked",
|
|
[
|
|
"bao write x token=ghp_abcdef0123456789abcdef0123", # github token prefix
|
|
"x=AKIAIOSFODNN7EXAMPLE", # aws key id
|
|
"header=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9", # jwt prefix
|
|
"val=ZmFrZXNlY3JldDEyMzQ1Njc4OWFiY2RlZmdoaWprbA", # high-entropy run
|
|
],
|
|
)
|
|
def test_handoff_secret_material_rejected(tmp_path, leaked):
|
|
bad = dict(ROUTED_ENTRY)
|
|
bad["fetch_command"] = leaked
|
|
with pytest.raises(CatalogError, match="secret|high-entropy"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_handoff_template_with_placeholders_accepted(tmp_path):
|
|
ok = dict(ROUTED_ENTRY)
|
|
ok["fetch_command"] = "bao kv get -field=<FIELD> platform/workloads/<domain>/<bundle>"
|
|
catalog = load_catalog(_write_catalog(tmp_path, [ok]))
|
|
assert catalog.get("openbao-api-key").fetch_command.startswith("bao kv get")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# find ranking
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_find_active_excludes_draft():
|
|
catalog = load_catalog(_repo_catalog())
|
|
ids = [e.id for e in catalog.find("s3 temporary credentials")]
|
|
assert "object-storage-sts" not in ids
|
|
|
|
|
|
def test_find_all_includes_draft():
|
|
catalog = load_catalog(_repo_catalog())
|
|
ids = [e.id for e in catalog.find("s3 temporary credentials", include_draft=True)]
|
|
assert "object-storage-sts" in ids
|
|
|
|
|
|
def test_find_issue_core_lane_active():
|
|
catalog = load_catalog(_repo_catalog())
|
|
ids = [e.id for e in catalog.find("issue core api key")]
|
|
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"
|
|
|
|
|
|
def test_find_openrouter_key():
|
|
catalog = load_catalog(_repo_catalog())
|
|
matches = catalog.find("openrouter api key", include_draft=True)
|
|
assert matches and matches[0].id == "openrouter-llm-connect"
|
|
|
|
|
|
def test_find_reuse_surface_hub_write_token():
|
|
catalog = load_catalog(_repo_catalog())
|
|
matches = catalog.find("REUSE_SURFACE_TOKEN", limit=3)
|
|
assert matches and matches[0].id == "reuse-surface-hub-write-token"
|
|
|
|
|
|
def test_reuse_surface_hub_write_token_lane_is_resolvable():
|
|
catalog = load_catalog(_repo_catalog())
|
|
e = catalog.get("reuse-surface-hub-write-token")
|
|
assert e is not None and e.is_active and e.exec_capable
|
|
assert e.resolvable is True
|
|
assert e.owner_repo == "railiance-platform"
|
|
assert "platform/workloads/reuse/reuse-surface/runtime-secrets" in e.fetch_command
|
|
|
|
|
|
def test_find_object_storage_sts():
|
|
catalog = load_catalog(_repo_catalog())
|
|
matches = catalog.find("s3 temporary credentials", include_draft=True)
|
|
assert matches and matches[0].id == "object-storage-sts"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Review staleness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_days_since_review():
|
|
assert days_since_review("2026-06-01", today=date(2026, 6, 24)) == 23
|
|
|
|
|
|
def test_is_review_stale_past_threshold():
|
|
assert is_review_stale("2026-01-01", threshold_days=90, today=date(2026, 6, 24))
|
|
|
|
|
|
def test_is_review_stale_within_threshold():
|
|
assert not is_review_stale("2026-06-01", threshold_days=90, today=date(2026, 6, 24))
|
|
|
|
|
|
def test_catalog_stale_filters_entries():
|
|
catalog = load_catalog(_repo_catalog())
|
|
stale = catalog.stale(threshold_days=0, today=date(2026, 6, 25))
|
|
assert stale
|
|
assert all(e.reviewed <= "2026-06-24" for e in stale)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 "object-storage-sts" not in ids
|
|
assert "issue-core-ingestion-api-key" 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 "object-storage-sts" 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 data["warden_role"] == "issue"
|
|
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
|
|
# exec_capable lane surfaces as an "assist" role so agents see it is proxyable.
|
|
assert data["warden_role"] == "assist"
|
|
assert data["exec_capable"] is True
|
|
assert "steps" not in data
|
|
assert "next_action" in data
|
|
assert "proxy" in data["next_action"]
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_cli_list_stale_json(repo_catalog_env):
|
|
result = runner.invoke(
|
|
app, ["route", "list", "--stale", "--stale-days", "1", "--json"]
|
|
)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.stdout)
|
|
assert data
|
|
assert all("days_since_review" in row for row in data)
|
|
assert all(row["stale_threshold_days"] == 1 for row in data)
|
|
|
|
|
|
def test_cli_list_stale_empty_with_high_threshold(repo_catalog_env):
|
|
result = runner.invoke(
|
|
app, ["route", "list", "--stale", "--stale-days", "9999"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "No stale" in result.output
|
|
|
|
|
|
def test_cli_find_openrouter_draft_only_with_all(repo_catalog_env):
|
|
result = runner.invoke(
|
|
app, ["route", "find", "openrouter api key", "--all", "--json"]
|
|
)
|
|
assert result.exit_code == 0
|
|
ids = [e["id"] for e in json.loads(result.stdout)]
|
|
assert "openrouter-llm-connect" 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}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rotation / re-establishment guidance registry (WARDEN-WP-0026 T06)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_every_active_vending_lane_has_rotation_guidance():
|
|
"""Coverage gate: an active lane that vends a secret must say how to renew it."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
missing = [e.id for e in catalog.entries if e.is_active and e.vends_secret and not e.has_rotation]
|
|
assert not missing, f"active vending lanes lacking rotation guidance: {missing}"
|
|
|
|
|
|
def test_scorecard_rotation_coverage_check_passes_on_repo_catalog():
|
|
result = check_catalog_rotation_coverage()
|
|
assert result.passed, result.detail
|
|
|
|
|
|
def test_non_vending_lanes_are_exempt_from_rotation():
|
|
"""SSH (issue), login, and pointer-only lanes carry no rotation block."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
assert catalog.get("ssh-cert-host-access").vends_secret is False # issue lane
|
|
assert catalog.get("key-cape-oidc-login").vends_secret is False # login lane
|
|
assert catalog.get("ops-bridge-tunnel").vends_secret is False # pointer only
|
|
|
|
|
|
def test_rotation_block_parses_fields():
|
|
catalog = load_catalog(_repo_catalog())
|
|
rot = catalog.get("forgejo-admin-api-token").rotation
|
|
assert rot is not None
|
|
assert rot.method in ("rotate", "re-establish")
|
|
assert rot.owner == "railiance-platform"
|
|
assert rot.steps and all(isinstance(s, str) for s in rot.steps)
|
|
|
|
|
|
def test_re_establish_method_on_backup_lane():
|
|
catalog = load_catalog(_repo_catalog())
|
|
rot = catalog.get("railiance-backup-offsite-lane").rotation
|
|
assert rot is not None and rot.method == "re-establish"
|
|
|
|
|
|
def test_invalid_rotation_method_rejected(tmp_path):
|
|
entry = dict(ROUTED_ENTRY, rotation={"method": "renew", "owner": "x", "steps": ["a"]})
|
|
with pytest.raises(CatalogError, match="rotation.method"):
|
|
load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, entry]))
|
|
|
|
|
|
def test_rotation_steps_screened_for_pasted_token(tmp_path):
|
|
"""A high-entropy pasted token in prose is rejected; ordinary prose is allowed."""
|
|
leak = dict(ROUTED_ENTRY, rotation={
|
|
"method": "rotate", "owner": "x",
|
|
"steps": ["set the value to ghp_" + "aB3dE5" * 6], # mixed alnum → high-entropy run
|
|
})
|
|
with pytest.raises(CatalogError, match="high-entropy|secret"):
|
|
load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, leak]))
|
|
|
|
|
|
def test_rotation_prose_allows_ordinary_sentences(tmp_path):
|
|
"""Words like 'exists.' must not trip the terse 's.' prefix screen."""
|
|
ok = dict(ROUTED_ENTRY, rotation={
|
|
"method": "rotate", "owner": "railiance-platform",
|
|
"steps": ["Rotate per the concrete workload's entry when one exists."],
|
|
})
|
|
catalog = load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, ok]))
|
|
assert catalog.get("openbao-api-key").rotation.steps
|
|
|
|
|
|
def test_rotate_guide_cli_json():
|
|
result = runner.invoke(app, ["rotate-guide", "forgejo-admin-api-token", "--json"])
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.stdout)
|
|
assert payload["method"] == "rotate"
|
|
assert payload["owner"] == "railiance-platform"
|
|
assert payload["steps"]
|
|
|
|
|
|
def test_rotate_guide_cli_ssh_lane_is_graceful():
|
|
# SSH renewal is re-issuance, not a static rotation — exit 0, not an error.
|
|
result = runner.invoke(app, ["rotate-guide", "ssh-cert-host-access"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent read-boundary + risk class (WARDEN-WP-0026 T04)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_high_risk_lanes_classified():
|
|
catalog = load_catalog(_repo_catalog())
|
|
high = {e.id for e in catalog.entries if e.is_high_risk}
|
|
assert "railiance-backup-offsite-lane" in high
|
|
assert "forgejo-admin-api-token" in high
|
|
assert "openrouter-llm-connect" in high
|
|
# WARDEN-WP-0033-T02: these two were asserted standard here, and the assertion
|
|
# held a defective grade still. Both paths carry a second credential the grade
|
|
# ignored -- GITEA_BACKEND_TOKEN (CCR-2026-0002) and the dual-consumer webhook
|
|
# HMAC (CCR-2026-0005). A read discloses every field at a path, so the grade
|
|
# must cover the union, not the headline field.
|
|
assert "issue-core-ingestion-api-key" in high
|
|
assert "reuse-surface-hub-write-token" in high
|
|
|
|
|
|
def test_invalid_risk_rejected(tmp_path):
|
|
bad = dict(ROUTED_ENTRY, risk="critical")
|
|
with pytest.raises(CatalogError, match="risk"):
|
|
load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, bad]))
|
|
|
|
|
|
def test_backup_lane_promoted_and_resolvable():
|
|
"""WP-0026 T07 — CCR-2026-0004 lane is active, resolvable, high-risk, has rotation."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
e = catalog.get("railiance-backup-offsite-lane")
|
|
assert e is not None
|
|
assert e.status == "active"
|
|
assert e.resolvable is True
|
|
assert e.is_high_risk is True
|
|
assert e.has_rotation is True
|
|
assert e.rotation.method == "re-establish"
|
|
assert "NC_WEBDAV_TOKEN" in (e.fetch_command or "")
|
|
assert "<" not in (e.fetch_command or "")
|
|
|
|
|
|
def test_route_show_json_includes_risk():
|
|
result = runner.invoke(app, ["route", "show", "railiance-backup-offsite-lane", "--json"])
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.stdout)
|
|
assert payload["risk"] == "high"
|
|
assert payload["high_risk"] is True
|
|
assert payload["resolvable"] is True
|
|
assert payload["status"] == "active"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delegation register (WARDEN-WP-0030)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_every_catalog_entry_declares_delegation():
|
|
catalog = load_catalog(_repo_catalog())
|
|
missing = [e.id for e in catalog.entries if e.delegation is None]
|
|
assert missing == [], f"entries missing delegation block: {missing}"
|
|
|
|
|
|
def test_every_proxy_declares_delegation():
|
|
"""A new exec_capable proxy cannot land without answering the ownership question."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
missing = [
|
|
e.id
|
|
for e in catalog.entries
|
|
if e.exec_capable and not e.warden_executes and e.delegation is None
|
|
]
|
|
assert missing == [], f"proxy lanes missing delegation: {missing}"
|
|
|
|
|
|
def test_ssh_lane_is_permanent_delegation():
|
|
e = load_catalog(_repo_catalog()).get("ssh-cert-host-access")
|
|
assert e.delegation is not None
|
|
assert e.delegation.mode == "permanent"
|
|
assert e.delegation.intended_owner is None
|
|
assert e.is_interim is False
|
|
|
|
|
|
def test_native_exec_lanes_are_native_delegation():
|
|
catalog = load_catalog(_repo_catalog())
|
|
for eid, owner in (
|
|
("whynot-design-npm-publish", "secrets-engine"),
|
|
("ops-warden-warden-sign-token", "railiance-platform"),
|
|
):
|
|
e = catalog.get(eid)
|
|
assert e.delegation is not None
|
|
assert e.delegation.mode == "native"
|
|
assert e.delegation.intended_owner == owner
|
|
|
|
|
|
def test_founder_interim_lanes_classified():
|
|
catalog = load_catalog(_repo_catalog())
|
|
expected = {
|
|
"rapp-qonto-keycape-client": "key-cape",
|
|
"binky-company-email-imap": "tenant-engine",
|
|
"binky-qonto-api": "tenant-engine",
|
|
"railiance-backup-offsite-lane": "railiance-platform",
|
|
"agent-harness-forgejo-deploy": "railiance-platform",
|
|
}
|
|
for eid, owner in expected.items():
|
|
e = catalog.get(eid)
|
|
assert e is not None and e.delegation is not None
|
|
assert e.delegation.mode == "interim"
|
|
assert e.delegation.intended_owner == owner
|
|
assert e.delegation.blocked_on
|
|
|
|
|
|
def test_missing_delegation_is_implicit_interim(tmp_path):
|
|
catalog = load_catalog(_write_catalog(tmp_path, [dict(ROUTED_ENTRY)]))
|
|
e = catalog.get("openbao-api-key")
|
|
d = e.effective_delegation
|
|
assert e.delegation is None
|
|
assert d.implicit is True
|
|
assert d.mode == "interim"
|
|
assert d.intended_owner is None
|
|
assert "unclassified" in (d.blocked_on or "")
|
|
|
|
|
|
def test_interim_without_blocked_on_rejected(tmp_path):
|
|
bad = dict(
|
|
ROUTED_ENTRY,
|
|
delegation={
|
|
"mode": "interim",
|
|
"intended_owner": "secrets-engine",
|
|
"reviewed": "2026-08-15",
|
|
},
|
|
)
|
|
with pytest.raises(CatalogError, match="blocked_on"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_non_permanent_without_owner_rejected(tmp_path):
|
|
bad = dict(
|
|
ROUTED_ENTRY,
|
|
delegation={"mode": "native", "reviewed": "2026-08-15"},
|
|
)
|
|
with pytest.raises(CatalogError, match="intended_owner"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_invalid_delegation_mode_rejected(tmp_path):
|
|
bad = dict(
|
|
ROUTED_ENTRY,
|
|
delegation={"mode": "maybe", "intended_owner": "x", "reviewed": "2026-08-15"},
|
|
)
|
|
with pytest.raises(CatalogError, match="delegation.mode"):
|
|
load_catalog(_write_catalog(tmp_path, [bad]))
|
|
|
|
|
|
def test_catalog_gaps_lists_only_interim():
|
|
catalog = load_catalog(_repo_catalog())
|
|
gap_ids = {e.id for e in catalog.gaps(include_draft=True)}
|
|
assert "ssh-cert-host-access" not in gap_ids
|
|
assert "whynot-design-npm-publish" not in gap_ids
|
|
assert "binky-company-email-imap" in gap_ids
|
|
# WARDEN-WP-0033: openbao-api-key was listed here as an interim cover. It is
|
|
# not one -- its path_template is a <domain>/<workload>/<bundle> routing
|
|
# pattern rather than a single secret lane, so there is no front door for
|
|
# anyone to take over. secrets-engine refused it on exactly that ground and
|
|
# ops-warden agrees. A pointer to OpenBao is not a gap ops-warden is holding,
|
|
# and counting it as one overstated the interim surface by a lane.
|
|
assert "openbao-api-key" not in gap_ids
|
|
assert all(catalog.get(i).is_interim for i in gap_ids)
|
|
|
|
|
|
def test_cli_route_gaps_json(repo_catalog_env):
|
|
result = runner.invoke(app, ["route", "gaps", "--json"])
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.stdout)
|
|
assert data
|
|
ids = {row["id"] for row in data}
|
|
assert "binky-company-email-imap" in ids
|
|
assert "ssh-cert-host-access" not in ids
|
|
for row in data:
|
|
assert row["mode"] == "interim"
|
|
assert "intended_owner" in row
|
|
assert "blocked_on" in row
|
|
assert "days_since_review" in row
|
|
|
|
|
|
def test_cli_route_show_includes_delegation(repo_catalog_env):
|
|
result = runner.invoke(app, ["route", "show", "binky-qonto-api", "--json"])
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.stdout)
|
|
assert data["delegation"]["mode"] == "interim"
|
|
assert data["delegation"]["intended_owner"] == "tenant-engine"
|
|
assert data["delegation"]["implicit"] is False
|
|
|
|
|
|
# --- ADR-0007: absence is not a grade (WARDEN-WP-0032-T06) ------------------
|
|
|
|
|
|
def _bare_entry(**overrides):
|
|
"""A minimal RouteEntry, so these tests exercise defaults and nothing else."""
|
|
from warden.routing.models import RouteEntry
|
|
|
|
fields = dict(
|
|
id="x",
|
|
title="t",
|
|
need_keywords=[],
|
|
owner_repo="r",
|
|
subsystem="s",
|
|
warden_executes=False,
|
|
wiki_ref="w",
|
|
canon_ref="c",
|
|
reviewed="2026-08-20",
|
|
status="active",
|
|
)
|
|
fields.update(overrides)
|
|
return RouteEntry(**fields)
|
|
|
|
|
|
def test_every_repo_catalog_lane_is_explicitly_graded():
|
|
"""The CI gate. A lane added without a `risk` grade is a defect (ADR-0007)."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
ungraded = sorted(e.id for e in catalog.entries if not e.is_graded)
|
|
assert ungraded == [], (
|
|
f"{len(ungraded)} catalog lane(s) carry no explicit risk grade: {ungraded}. "
|
|
"ADR-0007: absence is not a grade — grade the lane on merit."
|
|
)
|
|
|
|
|
|
def test_ungraded_lane_fails_safe_to_high_risk():
|
|
"""RISK-F-0003 regression: an omitted grade must not wave a lane through.
|
|
|
|
Before ADR-0007 the dataclass default was "standard", so a lane that simply
|
|
omitted the field landed outside the agent read-boundary silently.
|
|
"""
|
|
entry = _bare_entry()
|
|
assert entry.risk == "ungraded"
|
|
assert entry.is_graded is False
|
|
assert entry.is_high_risk is True
|
|
|
|
|
|
def test_unrecognised_grade_is_treated_as_high():
|
|
"""A grade from a newer catalog must not be read as permission."""
|
|
entry = _bare_entry(risk="spicy")
|
|
assert entry.is_high_risk is True
|
|
assert entry.is_graded is False
|
|
|
|
|
|
def test_ungraded_risk_uses_maturity_derived_zone_default():
|
|
entry = _bare_entry()
|
|
assert entry.risk_for_zone(
|
|
effective_zone="z0-experimental",
|
|
admission="satisfied",
|
|
synthetic_only=True,
|
|
) == "standard"
|
|
assert entry.risk_for_zone(
|
|
effective_zone="z0-experimental",
|
|
admission="unknown",
|
|
synthetic_only=True,
|
|
) == "high"
|
|
assert entry.risk_for_zone(
|
|
effective_zone="z3-critical",
|
|
admission="satisfied",
|
|
) == "critical"
|
|
assert entry.risk_for_zone(effective_zone="unknown") == "high"
|
|
|
|
|
|
def test_explicit_risk_grade_always_wins_over_zone_default():
|
|
entry = _bare_entry(risk="standard")
|
|
assert entry.risk_for_zone(
|
|
effective_zone="z3-critical", admission="satisfied"
|
|
) == "standard"
|
|
|
|
|
|
def test_low_risk_vocabulary_is_explicit():
|
|
for grade in ("standard", "low", "accepted"):
|
|
entry = _bare_entry(risk=grade)
|
|
assert entry.is_high_risk is False, grade
|
|
assert entry.is_graded is True, grade
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Blocker staleness cadence + verification (WARDEN-WP-0033-T05)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_blocker_cadence_is_separate_from_pointer_cadence():
|
|
"""Two claims with different half-lives must not share one threshold.
|
|
|
|
"Is this still the right owner and page?" is quarterly. "Has the owner
|
|
answered yet?" is not. Sharing 90 days made the second one inert -- the
|
|
register was six days old, so it could not have fired for months.
|
|
"""
|
|
from warden.routing.catalog import DEFAULT_BLOCKER_STALE_DAYS, DEFAULT_STALE_DAYS
|
|
|
|
assert DEFAULT_STALE_DAYS == 90
|
|
assert DEFAULT_BLOCKER_STALE_DAYS == 14
|
|
assert DEFAULT_BLOCKER_STALE_DAYS < DEFAULT_STALE_DAYS
|
|
|
|
|
|
def test_blocker_window_scales_with_risk_and_matches_risk_nexus():
|
|
"""risk-nexus stall windows: 14d critical/high, 30d medium, 60d low.
|
|
|
|
They offered the convention instead of a joint tool, so the two registers
|
|
agree only for as long as these numbers do.
|
|
"""
|
|
from warden.routing.catalog import blocker_stale_days
|
|
|
|
assert blocker_stale_days("high") == 14
|
|
assert blocker_stale_days("standard") == 30
|
|
assert blocker_stale_days("low") == 60
|
|
# An ungraded lane gets the SHORTEST window, not the longest -- ADR-0007 makes
|
|
# an absent grade a defect, so its blocker is the least trustworthy of all.
|
|
assert blocker_stale_days("ungraded") == 14
|
|
assert blocker_stale_days(None) == 14
|
|
# An explicit --stale-days still wins.
|
|
assert blocker_stale_days("low", 7) == 7
|
|
|
|
|
|
def test_asked_and_waiting_is_not_verification():
|
|
"""The failure this whole change exists to catch.
|
|
|
|
A lane asked today reads as reviewed today. The secrets-engine blocker sat
|
|
in exactly that state for ten days while looking current.
|
|
"""
|
|
from warden.routing.models import Delegation
|
|
|
|
asked = Delegation(mode="interim", intended_owner="x", blocked_on="y",
|
|
reviewed="2026-08-21", verified="asked-and-waiting")
|
|
assert asked.is_verified is False
|
|
|
|
for method in ("owner-confirmed", "source-read"):
|
|
d = Delegation(mode="interim", intended_owner="x", blocked_on="y",
|
|
reviewed="2026-08-21", verified=method)
|
|
assert d.is_verified is True, method
|
|
|
|
|
|
def test_stale_gaps_flags_unverified_even_when_the_date_is_today():
|
|
catalog = load_catalog(_repo_catalog())
|
|
stale = {e.id for e in catalog.stale_gaps(include_draft=True, today=date(2026, 8, 21))}
|
|
# Asked of key-cape on 2026-08-21 and unanswered -- zero days old, still stale.
|
|
assert "key-cape-oidc-login" in stale
|
|
# Confirmed by the owner the same day -- fresh.
|
|
assert "issue-core-ingestion-api-key" not in stale
|
|
|
|
|
|
def test_invalid_verification_method_rejected(tmp_path):
|
|
entry = dict(ROUTED_ENTRY)
|
|
entry["delegation"] = {
|
|
"mode": "interim", "intended_owner": "secrets-engine",
|
|
"blocked_on": "pending", "reviewed": "2026-08-21", "verified": "probably-fine",
|
|
}
|
|
with pytest.raises(CatalogError, match="verified"):
|
|
load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, entry]))
|
|
|
|
|
|
def test_every_interim_lane_records_how_it_was_verified():
|
|
"""Structural, not time-based, so it never fails on a calendar day alone."""
|
|
catalog = load_catalog(_repo_catalog())
|
|
missing = [
|
|
e.id for e in catalog.gaps(include_draft=True)
|
|
if e.effective_delegation.verified is None
|
|
]
|
|
assert not missing, f"interim lanes with no `verified`: {missing}"
|
|
|
|
|
|
def test_cli_route_gaps_fail_on_stale_exits_3(repo_catalog_env):
|
|
result = runner.invoke(app, ["route", "gaps", "--fail-on-stale", "--json"])
|
|
assert result.exit_code == 3
|
|
rows = json.loads(result.stdout)
|
|
assert any(r["stale"] for r in rows)
|
|
# An asked-and-waiting lane stays stale until it is verified, regardless of
|
|
# how many calendar days have elapsed since the request.
|
|
assert any(
|
|
r["stale"]
|
|
and r["verified"] == "asked-and-waiting"
|
|
for r in rows
|
|
)
|