tegwick
3a1bd4f1c8
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0217e-8c4c-7383-be6b-f50a6e485306
134 lines
5 KiB
Python
134 lines
5 KiB
Python
"""ops-warden routing contract.
|
|
|
|
ops-warden routes non-SSH credential needs here. It must NOT vend secret values.
|
|
A route result is a pointer: catalog id, readiness, decision status, and the safe
|
|
next command. This module computes that pointer for a lane. No value is read.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from secrets_engine.catalog import CatalogEntry
|
|
from secrets_engine.decisions import Decision, resolve_decision
|
|
from secrets_engine.errors import DecisionError
|
|
from secrets_engine.openbao import OpenBaoClient
|
|
|
|
|
|
@dataclass
|
|
class RouteResult:
|
|
catalog_id: str
|
|
kind: str
|
|
owner: str
|
|
stage: str
|
|
decision_status: str
|
|
decision_ref: str
|
|
review_url: str
|
|
metadata_applied: bool
|
|
value_present: bool
|
|
missing_fields: list[str]
|
|
ready: bool
|
|
next_command: str
|
|
missing: str
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def route_lane(
|
|
entry: CatalogEntry,
|
|
*,
|
|
hub_url: str,
|
|
repo_root: Path,
|
|
client: OpenBaoClient | None = None,
|
|
) -> RouteResult:
|
|
"""Build the front-door routing pointer for a lane. Never reads the value."""
|
|
decision_status = "n/a (bootstrap-only)"
|
|
decision_ref = entry.approval.get("decision_ref", "")
|
|
review_url = ""
|
|
decision: Decision | None = None
|
|
if entry.approval_required():
|
|
try:
|
|
decision = resolve_decision(
|
|
hub_url=hub_url, repo_root=repo_root, decision_ref=decision_ref
|
|
)
|
|
decision_status = decision.status
|
|
review_url = decision.review_url
|
|
except DecisionError:
|
|
decision_status = "missing"
|
|
|
|
metadata_applied = False
|
|
value_present = False
|
|
missing_fields: list[str] = list(entry.fields) if entry.stores_kv_value() else []
|
|
if client is not None and client.is_reachable():
|
|
if entry.has_delivery_auth:
|
|
policy_applied = client.read_policy(entry.policy_name) is not None
|
|
role_applied = client.approle_exists(entry.role_name)
|
|
metadata_applied = policy_applied and role_applied
|
|
if entry.stores_kv_value():
|
|
# One read produces booleans only; readiness requires every field.
|
|
presence = client.kv_fields_present(entry.mount, entry.path, entry.fields)
|
|
missing_fields = [field for field in entry.fields if not presence.get(field)]
|
|
value_present = bool(entry.fields) and not missing_fields
|
|
else:
|
|
# Auth-capability lanes have no stored value; a fresh secret_id is minted
|
|
# on demand through the handoff command once metadata exists.
|
|
value_present = metadata_applied
|
|
|
|
approved = not entry.approval_required() or (decision is not None and decision.is_approved())
|
|
ready = approved and metadata_applied and value_present
|
|
|
|
if not approved:
|
|
missing = f"approved decision for '{decision_ref}'"
|
|
next_command = f"secrets-engine decision inspect {decision_ref or entry.id}"
|
|
elif not metadata_applied:
|
|
if entry.kind == "kv" and entry.delivery_auth_management == "existing":
|
|
missing = "externally managed OpenBao policy/AppRole readiness"
|
|
next_command = (
|
|
f"secrets-engine verify {entry.id} --positive --negative"
|
|
)
|
|
elif entry.kind == "kv" and not entry.has_delivery_auth:
|
|
missing = "native delivery auth declaration"
|
|
next_command = f"secrets-engine plan {decision_ref or entry.id} --stage {entry.stage}"
|
|
else:
|
|
missing = "OpenBao policy/role apply"
|
|
next_command = f"secrets-engine apply {decision_ref or entry.id} --stage {entry.stage}"
|
|
elif entry.kind == "auth-capability":
|
|
missing = ""
|
|
next_command = (
|
|
f"secrets-engine handoff {entry.id} --stage {entry.stage} "
|
|
"--role-id-file <path> --secret-id-file <path>"
|
|
)
|
|
elif not value_present:
|
|
names = ", ".join(missing_fields)
|
|
missing = f"provisioned secret fields: {names}"
|
|
next_field = missing_fields[0] if missing_fields else entry.fields[0]
|
|
next_command = (
|
|
f"secrets-engine provision {entry.id} --stage {entry.stage} "
|
|
f"--field {next_field} --from-file <path>"
|
|
)
|
|
else:
|
|
missing = ""
|
|
if {"exec-env", "npm-config"}.intersection(entry.delivery_modes):
|
|
next_command = f"secrets-engine exec --catalog {entry.id} -- <command...>"
|
|
else:
|
|
next_command = (
|
|
f"secrets-engine verify {entry.id} --positive --negative"
|
|
)
|
|
|
|
return RouteResult(
|
|
catalog_id=entry.id,
|
|
kind=entry.kind,
|
|
owner=entry.owner,
|
|
stage=entry.stage,
|
|
decision_status=decision_status,
|
|
decision_ref=decision_ref,
|
|
review_url=review_url,
|
|
metadata_applied=metadata_applied,
|
|
value_present=value_present,
|
|
missing_fields=missing_fields,
|
|
ready=ready,
|
|
next_command=next_command,
|
|
missing=missing,
|
|
)
|