207 lines
6.9 KiB
Python
207 lines
6.9 KiB
Python
|
"""Read-only secret-use evidence surface for kings-guard.
|
||
|
|
|
||
|
|
This is not ``route`` or ``audit``. It never contacts OpenBao and never
|
||
|
|
returns secret material. Completeness is not claimed; missing local evidence
|
||
|
|
is not treated as non-occurrence.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
from secrets_engine.catalog import CatalogEntry, load_catalog
|
||
|
|
from secrets_engine.evidence_class import classification_path, load_classification_rules
|
||
|
|
from secrets_engine.redact import looks_secret, redact_text
|
||
|
|
|
||
|
|
LANE_FIELDS = (
|
||
|
|
"as_of",
|
||
|
|
"catalog_id",
|
||
|
|
"stage",
|
||
|
|
"kind",
|
||
|
|
"mount",
|
||
|
|
"path",
|
||
|
|
"field_names",
|
||
|
|
"ready",
|
||
|
|
"session_handle",
|
||
|
|
"revocation_attempted",
|
||
|
|
"revocation_succeeded",
|
||
|
|
"lifecycle_operation",
|
||
|
|
"decision_id",
|
||
|
|
"stance_stage",
|
||
|
|
"stance_failure_mode",
|
||
|
|
"evidence_kind",
|
||
|
|
)
|
||
|
|
LIFECYCLE_ACTIONS = {
|
||
|
|
"lifecycle-suspend": "suspend",
|
||
|
|
"lifecycle-deactivate": "deactivate",
|
||
|
|
"lifecycle-destroy": "destroy",
|
||
|
|
"revoke": "revoke",
|
||
|
|
"session-revoke": "revoke",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def declared_cadence() -> dict[str, str]:
|
||
|
|
"""Publish the load-bearing cadence already declared next to the bound."""
|
||
|
|
path = classification_path()
|
||
|
|
try:
|
||
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
|
|
except (OSError, yaml.YAMLError):
|
||
|
|
data = {}
|
||
|
|
raw = data.get("cadence") if isinstance(data, dict) else {}
|
||
|
|
if not isinstance(raw, dict):
|
||
|
|
raw = {}
|
||
|
|
cadence = {
|
||
|
|
"form": str(raw.get("load_bearing_form") or "heartbeat"),
|
||
|
|
"interval": str(raw.get("interval") or "1d"),
|
||
|
|
"command": str(raw.get("command") or "secrets-engine evidence heartbeat"),
|
||
|
|
}
|
||
|
|
# Keep the pin honest: classification rules must still load.
|
||
|
|
load_classification_rules()
|
||
|
|
return cadence
|
||
|
|
|
||
|
|
|
||
|
|
def _iter_records(evidence_dir: Path, catalog_id: str) -> list[dict[str, Any]]:
|
||
|
|
records: list[dict[str, Any]] = []
|
||
|
|
for path in sorted(Path(evidence_dir).glob("evidence-*.jsonl")):
|
||
|
|
try:
|
||
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
||
|
|
except OSError:
|
||
|
|
continue
|
||
|
|
for line in lines:
|
||
|
|
try:
|
||
|
|
record = json.loads(line)
|
||
|
|
except (json.JSONDecodeError, TypeError):
|
||
|
|
continue
|
||
|
|
if isinstance(record, dict) and record.get("catalog_id") == catalog_id:
|
||
|
|
records.append(record)
|
||
|
|
return records
|
||
|
|
|
||
|
|
|
||
|
|
def _last_with(records: list[dict[str, Any]], predicate) -> dict[str, Any] | None:
|
||
|
|
for record in reversed(records):
|
||
|
|
if predicate(record):
|
||
|
|
return record
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _safe_text(value: object) -> str:
|
||
|
|
if not isinstance(value, str) or not value:
|
||
|
|
return ""
|
||
|
|
return redact_text(value)
|
||
|
|
|
||
|
|
|
||
|
|
def snapshot_lane(
|
||
|
|
entry: CatalogEntry,
|
||
|
|
evidence_dir: Path,
|
||
|
|
*,
|
||
|
|
now: datetime | None = None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""One lane row. Catalog fields always; evidence fields only when present."""
|
||
|
|
as_of = (now or datetime.now(timezone.utc)).astimezone(timezone.utc).isoformat()
|
||
|
|
row: dict[str, Any] = {
|
||
|
|
"as_of": as_of,
|
||
|
|
"catalog_id": entry.id,
|
||
|
|
"stage": entry.stage,
|
||
|
|
"kind": entry.kind,
|
||
|
|
"mount": entry.mount,
|
||
|
|
"path": entry.path,
|
||
|
|
"field_names": list(entry.fields),
|
||
|
|
}
|
||
|
|
records = _iter_records(evidence_dir, entry.id)
|
||
|
|
verify = _last_with(
|
||
|
|
records, lambda rec: rec.get("action") == "verify" and rec.get("result") in {"pass", "verification-failed"}
|
||
|
|
)
|
||
|
|
if verify is not None:
|
||
|
|
row["ready"] = verify.get("result") == "pass"
|
||
|
|
|
||
|
|
def _session_bits(record: dict[str, Any]) -> None:
|
||
|
|
detail = record.get("detail") if isinstance(record.get("detail"), dict) else {}
|
||
|
|
session = detail.get("session") if isinstance(detail.get("session"), dict) else {}
|
||
|
|
handle = session.get("session_handle") or detail.get("session_handle") or detail.get("auth_session_handle")
|
||
|
|
if isinstance(handle, str) and handle and not looks_secret(handle):
|
||
|
|
row["session_handle"] = redact_text(handle)
|
||
|
|
if "revocation_attempted" in session or "auth_revocation_attempted" in detail:
|
||
|
|
row["revocation_attempted"] = bool(
|
||
|
|
session.get("revocation_attempted") or detail.get("auth_revocation_attempted")
|
||
|
|
)
|
||
|
|
if "revocation_succeeded" in session or "auth_revocation_succeeded" in detail:
|
||
|
|
row["revocation_succeeded"] = bool(
|
||
|
|
session.get("revocation_succeeded") or detail.get("auth_revocation_succeeded")
|
||
|
|
)
|
||
|
|
|
||
|
|
session_rec = _last_with(
|
||
|
|
records,
|
||
|
|
lambda rec: isinstance(rec.get("detail"), dict)
|
||
|
|
and (
|
||
|
|
isinstance(rec["detail"].get("session"), dict)
|
||
|
|
or "session_handle" in rec["detail"]
|
||
|
|
or "auth_session_handle" in rec["detail"]
|
||
|
|
),
|
||
|
|
)
|
||
|
|
if session_rec is not None:
|
||
|
|
_session_bits(session_rec)
|
||
|
|
|
||
|
|
life = _last_with(records, lambda rec: rec.get("action") in LIFECYCLE_ACTIONS)
|
||
|
|
if life is not None:
|
||
|
|
row["lifecycle_operation"] = LIFECYCLE_ACTIONS[str(life.get("action"))]
|
||
|
|
|
||
|
|
decided = _last_with(records, lambda rec: bool(rec.get("decision_id")))
|
||
|
|
if decided is not None:
|
||
|
|
decision_id = _safe_text(decided.get("decision_id"))
|
||
|
|
if decision_id:
|
||
|
|
row["decision_id"] = decision_id
|
||
|
|
|
||
|
|
stance = _last_with(
|
||
|
|
records,
|
||
|
|
lambda rec: isinstance(rec.get("detail"), dict)
|
||
|
|
and rec["detail"].get("stance_stage"),
|
||
|
|
)
|
||
|
|
if stance is not None:
|
||
|
|
detail = stance["detail"]
|
||
|
|
stage = _safe_text(detail.get("stance_stage"))
|
||
|
|
mode = _safe_text(detail.get("stance_failure_mode"))
|
||
|
|
if stage:
|
||
|
|
row["stance_stage"] = stage
|
||
|
|
if mode:
|
||
|
|
row["stance_failure_mode"] = mode
|
||
|
|
|
||
|
|
kind_rec = _last_with(records, lambda rec: rec.get("evidence_kind") in {"load-bearing", "attributive", "heartbeat"})
|
||
|
|
if kind_rec is not None:
|
||
|
|
row["evidence_kind"] = kind_rec["evidence_kind"]
|
||
|
|
|
||
|
|
extra = {key: row[key] for key in list(row) if key not in LANE_FIELDS}
|
||
|
|
for key in extra:
|
||
|
|
del row[key]
|
||
|
|
return row
|
||
|
|
|
||
|
|
|
||
|
|
def snapshot(
|
||
|
|
catalog_dir: Path,
|
||
|
|
evidence_dir: Path,
|
||
|
|
*,
|
||
|
|
catalog_id: str = "",
|
||
|
|
now: datetime | None = None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""Catalog snapshot for the secret-use surface. Never talks to OpenBao."""
|
||
|
|
current = now or datetime.now(timezone.utc)
|
||
|
|
entries = load_catalog(catalog_dir)
|
||
|
|
if catalog_id:
|
||
|
|
if catalog_id not in entries:
|
||
|
|
from secrets_engine.errors import CatalogError
|
||
|
|
|
||
|
|
raise CatalogError(f"unknown catalog id '{catalog_id}'")
|
||
|
|
chosen = [entries[catalog_id]]
|
||
|
|
else:
|
||
|
|
chosen = [entries[key] for key in sorted(entries)]
|
||
|
|
return {
|
||
|
|
"as_of": current.astimezone(timezone.utc).isoformat(),
|
||
|
|
"surface": "secret-use-evidence",
|
||
|
|
"completeness_claimed": False,
|
||
|
|
"cadence": declared_cadence(),
|
||
|
|
"lanes": [snapshot_lane(entry, evidence_dir, now=current) for entry in chosen],
|
||
|
|
}
|