coulomb/ops-warden
5
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions
ops-warden/src/warden/plan.py
tegwick 66db87e6ae feat: refuse to answer a write with a read (WARDEN-WP-0038) 
`warden plan` scored needs by keyword overlap with no notion of what the
caller wanted to DO, so "generate a successor secret and CAS-write it to two
custodians" matched the lane that READS that path and inherited its
`autonomous` verdict -- answered with --out/--exec/--wrap.

Two counterparties reported it in two days. key-cape distrusted the output on
principle and was right to; railiance-platform, answering as the write
authority being wrongly bypassed, said plainly that `founder_required` is the
verdict it should have returned and that until it is fixed a plan result must
not stand in for the owner's answer.

A mutating need on a lane ops-warden does not permanently own can no longer
reach any branch returning `autonomous`: it becomes `founder_required` with
an approve act naming the write owner, or `unroutable` with a CCR stub when
the lane admits no rotation route. Commands carry no read transport either
way, which is the half that made the wrong verdict actionable.

The ownership test does the work a verb list cannot. SSH certificate
issuance is itself a mutating act, so `delegation.mode: permanent` -- not the
absence of a verb -- separates ops-warden's own front door from someone
else's custody. A regression asserts `warden sign` still proceeds; a guard
that refused our own lane would be worse than the defect it fixes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013EPuTc18FjU5WFqoSEKH3C

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1276224@bnt-lap001
Assistant-Session: 426ec497-e1c4-4dd3-b417-dfce1ca1dbc3
2026-09-10 08:02:10 +02:00

553 lines
20 KiB
Python
Raw Blame History

"""Policy decision front door — ``warden plan`` (WARDEN-WP-0029 T01).
Composes the routing catalog, access handoff expansion, organization posture,
and flex-auth gate status into a typed verdict. Never holds secret values.
Does not re-implement keyword matching — delegates to ``Catalog.find``.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import List, Optional
from warden.access import expand_handoff, policy_gate_status
from warden.posture import PostureCatalog, load_posture
from warden.routing.catalog import Catalog, load_catalog
from warden.routing.models import RouteEntry
VERDICTS = ("autonomous", "founder_required", "unroutable")
FOUNDER_ACT_KINDS = ("oidc_login", "approve", "paste_once_provision")
_PROVISION_SIGNS = re.compile(
r"\b(provision|mint|onboard|paste|first[- ]time|rotate\s+into|put\s+into\s+openbao)\b"
r"|\bnew\b.{0,40}\b(secret|token|pat|key|credential)\b"
r"|\bstore\s+(?:the\s+)?(?:pat|token|key|secret)\b",
re.IGNORECASE,
)
_APPROVAL_SIGNS = re.compile(
r"\b(approv|red[- ]lane|ccr|policy\s+enable|prod\s+flip|break[- ]glass)\b",
re.IGNORECASE,
)
@dataclass
class FounderAct:
kind: str # oidc_login | approve | paste_once_provision
summary: str
details: dict = field(default_factory=dict)
def to_dict(self) -> dict:
return {"kind": self.kind, "summary": self.summary, "details": dict(self.details)}
@dataclass
class AccessPlan:
need: str
verdict: str
organization_posture: str
policy_gate: str
lane_id: Optional[str] = None
lane_title: Optional[str] = None
match_score: Optional[int] = None
commands: List[str] = field(default_factory=list)
founder_act: Optional[FounderAct] = None
ccr_stub: Optional[dict] = None
catalog: dict = field(default_factory=dict)
candidates: List[dict] = field(default_factory=list)
reasons: List[str] = field(default_factory=list)
actor: Optional[str] = None
domain: Optional[str] = None
def to_dict(self) -> dict:
return {
"need": self.need,
"verdict": self.verdict,
"organization_posture": self.organization_posture,
"policy_gate": self.policy_gate,
"lane_id": self.lane_id,
"lane_title": self.lane_title,
"match_score": self.match_score,
"commands": list(self.commands),
"founder_act": self.founder_act.to_dict() if self.founder_act else None,
"ccr_stub": self.ccr_stub,
"catalog": dict(self.catalog),
"candidates": list(self.candidates),
"reasons": list(self.reasons),
"actor": self.actor,
"domain": self.domain,
}
def _org_posture_id(posture: Optional[PostureCatalog]) -> str:
if posture is None:
return "unknown"
return posture.organization_posture.id
def _candidate_row(entry: RouteEntry, score: int) -> dict:
return {
"id": entry.id,
"title": entry.title,
"score": score,
"status": entry.status,
"resolvable": entry.resolvable,
"exec_capable": entry.exec_capable,
"warden_executes": entry.warden_executes,
"lane": entry.lane,
"risk": entry.risk,
}
#: Verbs that make a need a request to CHANGE custody rather than to read it
#: (WARDEN-WP-0038). Reported independently by key-cape (2026-09-08) and
#: railiance-platform (2026-09-09): a need saying "generate a successor secret and
#: CAS-write it to two custodians" scored well against the lane that READS that
#: path and inherited its `autonomous` verdict, answered with three read
#: transports.
#:
#: Deliberately not including "issue" or "sign". SSH certificate issuance is a
#: mutating act ops-warden owns outright, and the ownership test below is what
#: separates it from someone else's custody — not the absence of the verb.
MUTATE_TOKENS = frozenset({
"rotate", "rotating", "rotation", "rerotate",
"generate", "regenerate", "mint", "reissue",
"write", "rewrite", "put", "patch", "cas",
"provision", "reprovision", "create", "install",
"revoke", "revoking", "disable", "delete", "remove",
"replace", "successor", "reset", "restart", "update",
})
def _need_intent(need: str) -> str:
"""``mutate`` when the need asks to change a credential, else ``read``.
Token match, not substring: "update" must not fire on "updated docs" any
more than it already would, but "no-update" style hyphenation is normalised
the same way the scorer normalises it.
"""
tokens = {t.strip(".,;:()") for t in need.lower().replace("-", " ").split()}
return "mutate" if tokens & MUTATE_TOKENS else "read"
def _owns_write_authority(entry: RouteEntry) -> bool:
"""Whether ops-warden may itself perform a mutating act on this lane.
True only where ops-warden is the designed owner of the front door — the SSH
signing lane, `delegation.mode: permanent`. Everywhere else ops-warden is a
pointer or a caller-identity proxy (`ADR-0002`, `ADR-0005`), so a write is
another component's custody act and no plan verdict here can authorise it.
"""
return entry.effective_delegation.mode == "permanent"
def _score_for(catalog: Catalog, entry: RouteEntry, need: str) -> int:
if entry.id == need.strip():
return 100
tokens = [t for t in need.lower().replace("-", " ").split() if t]
return entry.match_score(tokens)
def _concrete(value: Optional[str]) -> bool:
"""True when a template has no ``<...>`` placeholders left."""
if not value:
return False
return "<" not in value and ">" not in value
def _lane_is_autonomous(entry: RouteEntry) -> bool:
"""Whether an agent can proceed without a founder act for this lane."""
if entry.warden_executes:
return True
if entry.lane == "login":
return False
if entry.resolvable:
return True
if entry.has_native_exec and _concrete(entry.exec_command):
return True
# Concrete owner fetch path (even if not exec_capable) — value already in custody
if _concrete(entry.fetch_command):
return True
# Pure pointer — follow wiki, no secret mechanics for founder
if not entry.has_handoff and not entry.exec_capable and not entry.has_native_exec:
return True
return False
def _autonomous_commands(entry: RouteEntry, domain: Optional[str]) -> List[str]:
cmds: List[str] = []
if entry.warden_executes:
if entry.cert_command:
cmds.append(entry.cert_command)
for step in entry.steps[:4]:
cmds.append(f"# {step}")
return cmds
expanded = expand_handoff(entry, domain=domain)
if entry.lane == "login":
cmds.append(
f"warden access {entry.id} --exec -- <reviewed-command>"
)
if expanded.fetch_command:
cmds.append(
"# owner login is contained by warden; do not invoke it separately"
)
return cmds
if entry.has_native_exec and entry.exec_command:
cmds.append(entry.exec_command)
if entry.pointer_command:
cmds.append(entry.pointer_command)
if entry.exec_capable:
base = f"warden access {entry.id}"
if domain:
base += f" --domain {domain}"
if entry.is_high_risk:
cmds.append(f"{base} --exec -- <cmd> # high-risk: no raw stdout")
cmds.append(f"{base} --out FILE")
cmds.append(f"{base} --wrap")
else:
cmds.append(f"{base} --fetch")
cmds.append(f"{base} --exec -- <cmd>")
if expanded.fetch_command:
cmds.append(f"# owner fetch (as you): {expanded.fetch_command}")
elif _concrete(expanded.fetch_command or entry.fetch_command):
cmds.append(expanded.fetch_command or entry.fetch_command or "")
if entry.wiki_ref:
cmds.append(f"# playbook: {entry.wiki_ref}")
elif entry.wiki_ref:
cmds.append(f"# follow owner playbook: {entry.wiki_ref}")
return [c for c in cmds if c]
def _founder_for_entry(entry: RouteEntry, need: str, domain: Optional[str]) -> FounderAct:
expanded = expand_handoff(entry, domain=domain)
if entry.lane == "login":
contained_command = (
f"warden access {entry.id} --exec -- <reviewed-command>"
)
return FounderAct(
kind="oidc_login",
summary=f"Interactive OIDC/MFA login via {entry.owner_repo}",
details={
"lane_id": entry.id,
"auth_method": expanded.auth_method,
"fetch_command": contained_command,
"desk_hint": (
"warden desk --from-plan (act=oidc_login); execute only through: "
+ contained_command
),
},
)
if entry.lane == "ceremony":
return FounderAct(
kind="approve",
summary=f"Attended owner ceremony approval required for {entry.id}",
details={
"lane_id": entry.id,
"wiki_ref": entry.wiki_ref,
"desk_hint": "warden desk --act approve --lane " + entry.id,
},
)
if _APPROVAL_SIGNS.search(need):
return FounderAct(
kind="approve",
summary=f"Founder approval required for {entry.id}",
details={
"lane_id": entry.id,
"wiki_ref": entry.wiki_ref,
"desk_hint": "warden desk --act approve --lane " + entry.id,
},
)
# Default founder path for non-resolvable secret lanes: paste-once provision
path = expanded.path_template or entry.path_template or "<openbao-path>"
return FounderAct(
kind="paste_once_provision",
summary=(
f"Provision secret value once into OpenBao path for {entry.id} "
"(no CLI paste; use warden desk)"
),
details={
"lane_id": entry.id,
"path_template": path,
"auth_method": expanded.auth_method,
"desk_hint": (
f"warden desk --act paste_once_provision --lane {entry.id}"
+ (f" --path {path}" if "<" not in (path or "") else "")
),
},
)
def _ccr_stub(need: str) -> dict:
return {
"title": f"CCR: new credential lane for {need[:80]}",
"status": "proposed",
"owner_hint": "railiance-platform (OpenBao) or owning subsystem",
"steps": [
"Draft CCR with path, policy, OIDC role, consumers",
"Add ops-warden catalog entry (pointers only; no secret values)",
"Playbook under wiki/playbooks/; promote status active when live",
],
"commands": [
"warden route list --all",
"# after CCR: edit registry/routing/catalog.yaml + playbook",
],
}
def build_plan(
need: str,
*,
actor: Optional[str] = None,
domain: Optional[str] = None,
catalog: Optional[Catalog] = None,
posture: Optional[PostureCatalog] = None,
include_draft: bool = False,
) -> AccessPlan:
"""Resolve *need* to a typed access plan. Pure of secret values."""
cat = catalog or load_catalog()
try:
post = posture if posture is not None else load_posture()
except Exception: # noqa: BLE001 — plan still works without posture file
post = None
gate = policy_gate_status()
org = _org_posture_id(post)
freshness = cat.freshness().to_dict()
raw_matches = cat.find(need, include_draft=include_draft, limit=8)
# Require score >= 2 (at least one full keyword hit). Score-1 hits are usually
# accidental substring overlaps (e.g. title word "or" inside an unrelated token).
scored = [(e, _score_for(cat, e, need)) for e in raw_matches]
matches = [(e, s) for e, s in scored if s >= 2]
candidates = [_candidate_row(e, s) for e, s in scored[:5]]
if not matches:
return AccessPlan(
need=need,
verdict="unroutable",
organization_posture=org,
policy_gate=gate,
ccr_stub=_ccr_stub(need),
catalog=freshness,
candidates=candidates,
reasons=["no catalog match for need (score < 2)"],
actor=actor,
domain=domain,
)
entry, score = matches[0]
intent = _need_intent(need)
# Draft-only top match without active alternatives → unroutable
if entry.status == "draft" and not include_draft:
return AccessPlan(
need=need,
verdict="unroutable",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
ccr_stub=_ccr_stub(need),
catalog=freshness,
candidates=candidates,
reasons=[f"top match {entry.id!r} is draft — promote or request CCR"],
actor=actor,
domain=domain,
)
# Login and ceremony lanes always need a human act. A ceremony is a pure
# owner pointer: it must never fall through to secret paste-once mechanics.
if entry.lane in ("login", "ceremony"):
act = _founder_for_entry(entry, need, domain)
return AccessPlan(
need=need,
verdict="founder_required",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=_autonomous_commands(entry, domain),
founder_act=act,
catalog=freshness,
candidates=candidates,
reasons=[
"login lane requires interactive founder/operator identity act"
if entry.lane == "login"
else "ceremony lane requires attended owner approval"
],
actor=actor,
domain=domain,
)
# Explicit approval language
if _APPROVAL_SIGNS.search(need) and not entry.warden_executes:
act = _founder_for_entry(entry, need, domain)
act.kind = "approve"
return AccessPlan(
need=need,
verdict="founder_required",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
founder_act=act,
catalog=freshness,
candidates=candidates,
reasons=["need text requests founder approval"],
actor=actor,
domain=domain,
)
# Explicit first-time provision language wins even if a concrete lane matched
if _PROVISION_SIGNS.search(need) and entry.lane == "secret" and not entry.warden_executes:
act = _founder_for_entry(entry, need, domain)
return AccessPlan(
need=need,
verdict="founder_required",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=[],
founder_act=act,
catalog=freshness,
candidates=candidates,
reasons=["need requires first-time provision — one founder act via warden desk"],
actor=actor,
domain=domain,
)
# --- mutating need on a lane ops-warden does not own (WARDEN-WP-0038) ------
#
# Must run before any branch that can return `autonomous`. A write is a
# custody act belonging to the lane owner, and `autonomous` is documented as
# the signal to proceed without the founder — so returning it here would let
# this front door authorise a mutation on someone else's custody. It would
# also answer a write with `--out`/`--exec`/`--wrap`, which are reads.
if intent == "mutate" and not _owns_write_authority(entry):
rotation = entry.rotation
owner = (rotation.owner if rotation else None) or entry.owner_repo
reasons = [
"need asks to change a credential, not read one",
f"write authority on this lane belongs to {owner}, not ops-warden",
]
if rotation is not None:
act = FounderAct(
kind="approve",
summary=(
f"Attended owner act required to change {entry.id} — "
f"{owner} holds write authority"
),
details={
"lane_id": entry.id,
"write_owner": owner,
"rotation_method": rotation.method,
"rotation_automatable": rotation.automatable,
"wiki_ref": entry.wiki_ref,
"guidance_command": f"warden rotate-guide {entry.id}",
"desk_hint": f"warden desk --act approve --lane {entry.id}",
},
)
if not rotation.automatable:
reasons.append(
"lane records the rotation as not automatable — an executable "
"driver must not attempt it"
)
return AccessPlan(
need=need,
verdict="founder_required",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
# Deliberately empty: every command this lane offers is a READ,
# and offering one against a write need is the reported defect.
commands=[f"warden rotate-guide {entry.id} # guidance, not execution"],
founder_act=act,
catalog=freshness,
candidates=candidates,
reasons=reasons,
actor=actor,
domain=domain,
)
stub = _ccr_stub(need)
stub["owner_hint"] = f"{owner} (write authority for {entry.id})"
reasons.append("lane records no rotation route — the act has no admitted transport")
return AccessPlan(
need=need,
verdict="unroutable",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=[],
ccr_stub=stub,
catalog=freshness,
candidates=candidates,
reasons=reasons,
actor=actor,
domain=domain,
)
if _lane_is_autonomous(entry):
return AccessPlan(
need=need,
verdict="autonomous",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=_autonomous_commands(entry, domain),
catalog=freshness,
candidates=candidates,
reasons=["lane is usable under current catalog without founder mechanics"],
actor=actor,
domain=domain,
)
# Template / non-concrete secret handoff → founder paste-once
if entry.lane == "secret":
act = _founder_for_entry(entry, need, domain)
return AccessPlan(
need=need,
verdict="founder_required",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=[],
founder_act=act,
catalog=freshness,
candidates=candidates,
reasons=[
"lane handoff still has placeholders or needs provision — "
"one founder act via warden desk"
],
actor=actor,
domain=domain,
)
# Fallback: autonomous with best-effort commands
return AccessPlan(
need=need,
verdict="autonomous",
organization_posture=org,
policy_gate=gate,
lane_id=entry.id,
lane_title=entry.title,
match_score=score,
commands=_autonomous_commands(entry, domain),
catalog=freshness,
candidates=candidates,
reasons=["matched lane; proceed via catalog handoff"],
actor=actor,
domain=domain,
)
Reference in a new issue View git blame Copy permalink