WARDEN-WP-0026 T06: rotation guidance registry + warden rotate-guide
- routing model: RotationGuide (method rotate|re-establish, steps, owner, automatable), RouteEntry.rotation + has_rotation + vends_secret. - catalog parser: validate rotation block; secret-material screen gains a prose-safe mode (high-entropy detector only) so authored steps aren't tripped by substrings like "s."/"exists.". - CLI: `warden rotate-guide <id>` (human + --json); route show --json now carries has_rotation + rotation. - scorecard: catalog_rotation_coverage — every active secret-vending lane must carry a rotation block (SSH/login/pointer lanes exempt). Promotion checklist criterion 9. - data: rotation blocks for all 7 active vending lanes + the draft railiance-backup lane (re-establish: age keypair regen + re-encrypt). - fix pre-existing collision: bare `npm` keyword on forgejo-admin -> forgejo-npm so "npm token" routes to the generic lane (restores test_access expectations). - tests: rotation parse/coverage/prose-screen/CLI in tests/test_routing.py; scorecard count 6 -> 7. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
ac09f21ad3
commit
c3eb59ea04
9 changed files with 385 additions and 12 deletions
|
|
@ -630,6 +630,21 @@ def _entry_summary(entry) -> dict:
|
|||
"canon_ref": entry.canon_ref,
|
||||
"reviewed": entry.reviewed,
|
||||
"status": entry.status,
|
||||
# Renewal guidance (WP-0026 T06) — advisory, no secret values. `has_rotation`
|
||||
# lets a caller gate before asking for the full block via `warden rotate-guide`.
|
||||
"has_rotation": entry.has_rotation,
|
||||
**(
|
||||
{
|
||||
"rotation": {
|
||||
"method": entry.rotation.method,
|
||||
"owner": entry.rotation.owner,
|
||||
"automatable": entry.rotation.automatable,
|
||||
"steps": entry.rotation.steps,
|
||||
}
|
||||
}
|
||||
if entry.has_rotation
|
||||
else {}
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -788,6 +803,63 @@ def route_show(
|
|||
)
|
||||
|
||||
|
||||
@app.command("rotate-guide")
|
||||
def rotate_guide(
|
||||
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 how to rotate or re-establish a lane's credential (WP-0026 T06).
|
||||
|
||||
Advisory renewal guidance held in the ops-warden registry — never a secret
|
||||
value, and ops-warden does not execute it (that is Strand B, WARDEN-WP-0027).
|
||||
"""
|
||||
catalog = _load_catalog()
|
||||
entry = catalog.get(entry_id)
|
||||
if entry is None:
|
||||
err.print(
|
||||
f"[red]Unknown routing id {entry_id!r}.[/red] Try: warden route find {entry_id!r}"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
if not entry.has_rotation:
|
||||
if output_json:
|
||||
print(json.dumps({"id": entry.id, "has_rotation": False}, indent=2))
|
||||
else:
|
||||
err.print(
|
||||
f"[yellow]No rotation guidance for {entry.id!r}.[/yellow] "
|
||||
+ (
|
||||
"This is the SSH lane — renewal is re-issuance (`warden sign`)."
|
||||
if entry.warden_executes
|
||||
else "Add a `rotation:` block to the catalog entry (WP-0026 T06)."
|
||||
)
|
||||
)
|
||||
raise typer.Exit(0 if entry.warden_executes else 1)
|
||||
|
||||
rot = entry.rotation
|
||||
if output_json:
|
||||
print(json.dumps(
|
||||
{
|
||||
"id": entry.id,
|
||||
"method": rot.method,
|
||||
"owner": rot.owner,
|
||||
"automatable": rot.automatable,
|
||||
"steps": rot.steps,
|
||||
},
|
||||
indent=2,
|
||||
))
|
||||
return
|
||||
|
||||
console.print(f"[bold]Rotation guidance — {entry.title}[/bold] ([cyan]{entry.id}[/cyan])")
|
||||
console.print(f" method : {rot.method} owner: {rot.owner} automatable: {rot.automatable}")
|
||||
console.print(" steps:")
|
||||
for i, step in enumerate(rot.steps, 1):
|
||||
console.print(f" {i}. {step}")
|
||||
console.print(
|
||||
"\n[dim]Advisory only — ops-warden holds no value and does not execute this "
|
||||
"(one-command rotation is Strand B, WARDEN-WP-0027).[/dim]"
|
||||
)
|
||||
|
||||
|
||||
@route_app.command("find")
|
||||
def route_find(
|
||||
query: Annotated[str, typer.Argument(help="Free-text need, e.g. 'issue core api key'")],
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ from typing import List, Optional
|
|||
|
||||
import yaml
|
||||
|
||||
from warden.routing.models import RouteEntry
|
||||
from warden.routing.models import RotationGuide, RouteEntry
|
||||
|
||||
# Structured handoff string fields (WP-0014) — templates and pointers only.
|
||||
# Every one is scanned for accidental secret material; see _assert_no_secret_material.
|
||||
|
|
@ -60,6 +60,7 @@ _REQUIRED_FIELDS = (
|
|||
)
|
||||
_VALID_STATUS = ("active", "draft")
|
||||
_VALID_LANES = ("secret", "login")
|
||||
_VALID_ROTATION_METHODS = ("rotate", "re-establish")
|
||||
|
||||
# Default review cadence — see wiki/AccessRouting.md#drift-review-cadence
|
||||
DEFAULT_STALE_DAYS = 90
|
||||
|
|
@ -161,7 +162,9 @@ class Catalog:
|
|||
]
|
||||
|
||||
|
||||
def _assert_no_secret_material(entry_id: str, field_name: str, value: str) -> None:
|
||||
def _assert_no_secret_material(
|
||||
entry_id: str, field_name: str, value: str, *, prose: bool = False
|
||||
) -> None:
|
||||
"""Reject a handoff field that appears to embed a literal secret value.
|
||||
|
||||
The structured handoff fields are command/path *templates*: concrete values
|
||||
|
|
@ -169,15 +172,22 @@ def _assert_no_secret_material(entry_id: str, field_name: str, value: str) -> No
|
|||
catalog is git-tracked and agent-visible, so a leaked value here is the exact
|
||||
custody failure WP-0014 forbids. We screen for known token prefixes and for a
|
||||
long high-entropy run that is not a placeholder.
|
||||
|
||||
``prose=True`` (rotation guidance steps, WP-0026 T06) skips the *substring*
|
||||
prefix screen — short prefixes like ``s.`` or ``eyJ`` collide with ordinary
|
||||
English ("exists.", "artifacts.") — and relies on the high-entropy-run detector,
|
||||
which catches an actually-pasted token (a real ``hvs.``/``ghp_``/``sk-`` value
|
||||
carries a long high-entropy tail) while allowing plain sentences.
|
||||
"""
|
||||
lowered = value.lower()
|
||||
for prefix in _SECRET_PREFIXES:
|
||||
if prefix.lower() in lowered:
|
||||
raise CatalogError(
|
||||
f"entry {entry_id!r} field {field_name!r} appears to contain a literal "
|
||||
f"secret (matched {prefix!r}). Handoff fields are templates — use "
|
||||
"placeholders like <FIELD>/<PATH>, never a real value."
|
||||
)
|
||||
if not prose:
|
||||
for prefix in _SECRET_PREFIXES:
|
||||
if prefix.lower() in lowered:
|
||||
raise CatalogError(
|
||||
f"entry {entry_id!r} field {field_name!r} appears to contain a literal "
|
||||
f"secret (matched {prefix!r}). Handoff fields are templates — use "
|
||||
"placeholders like <FIELD>/<PATH>, never a real value."
|
||||
)
|
||||
for run in _HIGH_ENTROPY_RUN.findall(value):
|
||||
# Allow long placeholder/path/identifier tokens; flag anything else.
|
||||
if "<" in run or ">" in run:
|
||||
|
|
@ -190,6 +200,49 @@ def _assert_no_secret_material(entry_id: str, field_name: str, value: str) -> No
|
|||
)
|
||||
|
||||
|
||||
def _parse_rotation(entry_id: str, raw: Optional[dict]) -> Optional[RotationGuide]:
|
||||
"""Parse and validate an optional ``rotation:`` block (WP-0026 T06).
|
||||
|
||||
Advisory renewal guidance only — screened for secret material like every other
|
||||
catalog string. ``method`` must be rotate | re-establish; ``steps`` a non-empty
|
||||
list; ``owner`` required.
|
||||
"""
|
||||
if raw is None:
|
||||
return None
|
||||
if not isinstance(raw, dict):
|
||||
raise CatalogError(f"entry {entry_id!r} `rotation` must be a mapping")
|
||||
|
||||
method = str(raw.get("method", "")).strip()
|
||||
if method not in _VALID_ROTATION_METHODS:
|
||||
raise CatalogError(
|
||||
f"entry {entry_id!r} rotation.method {method!r} invalid "
|
||||
f"(expected one of {_VALID_ROTATION_METHODS})"
|
||||
)
|
||||
|
||||
steps_raw = raw.get("steps")
|
||||
if not isinstance(steps_raw, list) or not steps_raw:
|
||||
raise CatalogError(
|
||||
f"entry {entry_id!r} rotation.steps must be a non-empty list of steps"
|
||||
)
|
||||
steps = [str(s) for s in steps_raw]
|
||||
|
||||
owner = str(raw.get("owner", "")).strip()
|
||||
if not owner:
|
||||
raise CatalogError(f"entry {entry_id!r} rotation.owner is required")
|
||||
|
||||
# Screen advisory prose for accidental secret material (git-tracked, agent-visible).
|
||||
for i, step in enumerate(steps):
|
||||
_assert_no_secret_material(entry_id, f"rotation.steps[{i}]", step, prose=True)
|
||||
_assert_no_secret_material(entry_id, "rotation.owner", owner, prose=True)
|
||||
|
||||
return RotationGuide(
|
||||
method=method,
|
||||
steps=steps,
|
||||
owner=owner,
|
||||
automatable=bool(raw.get("automatable", False)),
|
||||
)
|
||||
|
||||
|
||||
def _parse_entry(raw: dict, index: int) -> RouteEntry:
|
||||
if not isinstance(raw, dict):
|
||||
raise CatalogError(f"entry #{index} is not a mapping")
|
||||
|
|
@ -272,6 +325,7 @@ def _parse_entry(raw: dict, index: int) -> RouteEntry:
|
|||
exec_owner=str(raw["exec_owner"]) if raw.get("exec_owner") else None,
|
||||
exec_command=handoff["exec_command"],
|
||||
pointer_command=handoff["pointer_command"],
|
||||
rotation=_parse_rotation(entry_id, raw.get("rotation")),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -11,6 +11,23 @@ from dataclasses import dataclass, field
|
|||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class RotationGuide:
|
||||
"""Structured-but-advisory renewal guidance for a lane (WARDEN-WP-0026 T06).
|
||||
|
||||
Held in the ops-warden registry, never in OpenBao. ``steps`` are authored
|
||||
advisory prose (screened for secret material like every catalog string) — they
|
||||
tell an operator *how* to renew, they are not executed here. ``method`` is
|
||||
``rotate`` (provider re-mints the same kind of credential) or ``re-establish``
|
||||
(regenerate from source, e.g. a new age keypair + re-encrypt). ``automatable``
|
||||
is a hint for a future Strand-B executable driver (WARDEN-WP-0027).
|
||||
"""
|
||||
method: str # "rotate" | "re-establish"
|
||||
steps: List[str]
|
||||
owner: str
|
||||
automatable: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class RouteEntry:
|
||||
id: str
|
||||
|
|
@ -50,11 +67,31 @@ class RouteEntry:
|
|||
exec_owner: Optional[str] = None # subsystem owning the native exec (e.g. secrets-engine)
|
||||
exec_command: Optional[str] = None # e.g. "secrets-engine exec --catalog <id> -- <cmd>"
|
||||
pointer_command: Optional[str] = None # e.g. "secrets-engine route <id> --json"
|
||||
# Rotation / re-establishment guidance (WP-0026 T06) — advisory, no secret values.
|
||||
rotation: Optional[RotationGuide] = None
|
||||
|
||||
@property
|
||||
def is_active(self) -> bool:
|
||||
return self.status == "active"
|
||||
|
||||
@property
|
||||
def has_rotation(self) -> bool:
|
||||
"""True when this lane carries renewal guidance (WP-0026 T06)."""
|
||||
return self.rotation is not None
|
||||
|
||||
@property
|
||||
def vends_secret(self) -> bool:
|
||||
"""True when this lane hands back a rotatable static secret value.
|
||||
|
||||
Rotation guidance (WP-0026 T06) applies to these. It excludes the SSH lane
|
||||
(short-lived certs — renewal is re-issuance), ``login`` lanes (re-auth, no
|
||||
stored value), and pure routing pointers with no secret path (tunnel,
|
||||
principals, emission sinks, policy checks).
|
||||
"""
|
||||
if self.warden_executes or self.lane != "secret":
|
||||
return False
|
||||
return bool(self.path_template or self.fetch_command or self.exec_owner)
|
||||
|
||||
@property
|
||||
def has_native_exec(self) -> bool:
|
||||
"""True when an owner-native exec front door is the primary path for this lane."""
|
||||
|
|
|
|||
|
|
@ -152,6 +152,43 @@ def check_file_permissions(state_dir: Path) -> CheckResult:
|
|||
)
|
||||
|
||||
|
||||
def check_catalog_rotation_coverage() -> CheckResult:
|
||||
"""Every active secret-vending catalog lane must carry rotation guidance (T06).
|
||||
|
||||
A lane an operator can obtain a *secret value* through must also tell them how
|
||||
to renew or re-establish it. Scoped to ``vends_secret`` lanes: this exempts the
|
||||
SSH lane (short-lived certs — renewal is re-issuance), ``login`` lanes (re-auth,
|
||||
no stored value), and pure routing pointers (tunnel, principals, emission
|
||||
sinks, policy checks) with no secret to rotate. Draft lanes are exempt until
|
||||
promoted.
|
||||
"""
|
||||
try:
|
||||
from warden.routing import load_catalog
|
||||
catalog = load_catalog()
|
||||
except Exception as e: # noqa: BLE001 — catalog missing/invalid is its own signal
|
||||
return CheckResult(
|
||||
name="catalog_rotation_coverage",
|
||||
passed=False,
|
||||
detail=f"could not load routing catalog: {e}",
|
||||
)
|
||||
|
||||
missing = [
|
||||
e.id
|
||||
for e in catalog.entries
|
||||
if e.is_active and e.vends_secret and not e.has_rotation
|
||||
]
|
||||
return CheckResult(
|
||||
name="catalog_rotation_coverage",
|
||||
passed=len(missing) == 0,
|
||||
detail=(
|
||||
f"active lanes lacking rotation guidance: {missing} — add a `rotation:` "
|
||||
"block (see WARDEN-WP-0026 T06)"
|
||||
if missing
|
||||
else "all active lanes carry rotation guidance"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def run_scorecard(state_dir: Path, inventory: PrincipalsInventory) -> List[CheckResult]:
|
||||
"""Run all cert-side scorecard checks. Returns list of CheckResult."""
|
||||
return [
|
||||
|
|
@ -161,4 +198,5 @@ def run_scorecard(state_dir: Path, inventory: PrincipalsInventory) -> List[Check
|
|||
check_no_stale_certs(state_dir),
|
||||
check_ttl_policy(state_dir, inventory),
|
||||
check_file_permissions(state_dir),
|
||||
check_catalog_rotation_coverage(),
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue