tegwick
5149946a4c
Ship posture-aware access planning: organization_posture=build (axis C), catalog freshness warnings, warden plan verdicts, localhost founder desk, and playbook/agent guidance that retire /tmp file-drop patterns. Compose route catalog + handoff rather than a second routing layer.
406 lines
14 KiB
Python
406 lines
14 KiB
Python
"""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,
|
|
}
|
|
|
|
|
|
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.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":
|
|
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": expanded.fetch_command,
|
|
"desk_hint": f"warden desk --from-plan (act=oidc_login) or: {expanded.fetch_command}",
|
|
},
|
|
)
|
|
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]
|
|
|
|
# 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 lanes always need a human identity act
|
|
if entry.lane == "login":
|
|
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"],
|
|
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,
|
|
)
|
|
|
|
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,
|
|
)
|