tegwick
b971403dad
Promote railiance-backup-offsite-lane to active/resolvable after capabilities-safe re-verify. Add catalog risk=high, agent read-boundary (exit 7 + OpenBao policy companion), EXPOSED taint via warden taint, and close WP-0026.
577 lines
22 KiB
Python
577 lines
22 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 pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from typer.testing import CliRunner
|
|
|
|
from warden.cli import app
|
|
from datetime import date
|
|
|
|
from warden.routing import CatalogError, load_catalog
|
|
from warden.routing.catalog import days_since_review, find_catalog_path, is_review_stale
|
|
|
|
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 <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",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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_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_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_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)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from warden.scorecard import check_catalog_rotation_coverage
|
|
|
|
|
|
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
|
|
# Ordinary workload secrets stay standard unless reclassified.
|
|
assert catalog.get("issue-core-ingestion-api-key").is_high_risk is False
|
|
|
|
|
|
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"
|