tegwick
36b707f0c3
T02 inbox check, wired into make check and verified against the actual 2026-08-19 failure — replayed at that moment it surfaces all three messages that were already waiting. T03 sweeps the rest of the quietly-tolerated class: bad dates, cadence off the ladder, undefined disclosure states, dangling constraint_on and related refs, embargoes without conditions, escalations without triggers. T04 requests verification of user-engine's tenant boundary — the first walk down the on-request path, chosen as a consumer not already known to fail it. T05 established by trying what this register can verify: cluster yes, OpenBao 403. T06 puts regulatory records on the findings ladder. T01 stays in progress: the procedure, make due and make checked exist, but arming something that runs them on schedule is a standing compute commitment and the operator's to make. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""Shared front-matter reading for the register tools.
|
|
|
|
The finding files are the source of truth. Nothing here writes to them.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import pathlib
|
|
import yaml
|
|
|
|
REPO = pathlib.Path(__file__).resolve().parent.parent
|
|
FINDINGS = REPO / "findings"
|
|
NOTES = REPO / "notes"
|
|
|
|
SEVERITIES = ["critical", "high", "medium", "low"]
|
|
GRADED_FIELDS = ["severity", "disclosure", "escalation", "next_check", "cadence"]
|
|
# A finding leaves the watch list only by being genuinely finished. Anything
|
|
# else — including a word the register has not seen before — stays watched, and
|
|
# the unknown word is reported. A finding must never fall out of the nag because
|
|
# somebody used a status the tooling did not recognise.
|
|
CLOSED_STATUSES = ("fixed", "withdrawn")
|
|
KNOWN_STATUSES = ("open", "accepted", "mitigated") + CLOSED_STATUSES
|
|
|
|
|
|
def watched(status: str | None) -> bool:
|
|
return status not in CLOSED_STATUSES
|
|
|
|
# The adaptive cadence ladder (operator ruling, 2026-08-20). A clean check
|
|
# climbs one rung; anything wrong drops straight back to `instant`. The rung a
|
|
# finding sits on is itself the signal: how stable this matter has been.
|
|
CADENCE = [
|
|
("instant", dt.timedelta(0)),
|
|
("1h", dt.timedelta(hours=1)),
|
|
("8h", dt.timedelta(hours=8)),
|
|
("24h", dt.timedelta(hours=24)),
|
|
("48h", dt.timedelta(hours=48)),
|
|
("96h", dt.timedelta(hours=96)),
|
|
("7d", dt.timedelta(days=7)),
|
|
("14d", dt.timedelta(days=14)),
|
|
("1mo", dt.timedelta(days=30)),
|
|
("1q", dt.timedelta(days=90)),
|
|
]
|
|
CADENCE_NAMES = [name for name, _ in CADENCE]
|
|
TOP_RUNG = CADENCE_NAMES[-1]
|
|
|
|
|
|
def rung_index(name: str) -> int:
|
|
return CADENCE_NAMES.index(name) if name in CADENCE_NAMES else 0
|
|
|
|
|
|
def interval(name: str) -> dt.timedelta:
|
|
return CADENCE[rung_index(name)][1]
|
|
|
|
|
|
def climb(name: str) -> str:
|
|
"""One clean check: up one rung, never past the quarter."""
|
|
return CADENCE_NAMES[min(rung_index(name) + 1, len(CADENCE) - 1)]
|
|
|
|
|
|
def reset() -> str:
|
|
"""Anything wrong: back to the bottom."""
|
|
return CADENCE_NAMES[0]
|
|
|
|
|
|
def load(path: pathlib.Path) -> dict:
|
|
text = path.read_text(encoding="utf-8")
|
|
if not text.startswith("---\n"):
|
|
raise ValueError(f"{path.name}: no front-matter")
|
|
_, fm, _body = text.split("---\n", 2)
|
|
data = yaml.safe_load(fm) or {}
|
|
data["_path"] = path
|
|
return data
|
|
|
|
|
|
def findings() -> list[dict]:
|
|
items = [load(p) for p in sorted(FINDINGS.glob("RISK-F-*.md"))]
|
|
return sorted(items, key=lambda f: f["id"], reverse=True)
|
|
|
|
|
|
def regulatory() -> list[dict]:
|
|
"""Regulatory records expire, so they ride the same ladder as findings."""
|
|
d = REPO / "docs" / "regulatory"
|
|
if not d.exists():
|
|
return []
|
|
return sorted((load(p) for p in d.glob("*.md") if p.name != "README.md"),
|
|
key=lambda r: r.get("id", ""))
|
|
|
|
|
|
def notes() -> list[dict]:
|
|
if not NOTES.exists():
|
|
return []
|
|
return sorted((load(p) for p in NOTES.glob("RISK-N-*.md")), key=lambda n: n["id"], reverse=True)
|
|
|
|
|
|
def moment(value) -> dt.datetime | None:
|
|
"""Parse a date or datetime front-matter value as UTC."""
|
|
if isinstance(value, dt.datetime):
|
|
return value if value.tzinfo else value.replace(tzinfo=dt.timezone.utc)
|
|
if isinstance(value, dt.date):
|
|
return dt.datetime.combine(value, dt.time(0, 0), tzinfo=dt.timezone.utc)
|
|
if isinstance(value, str) and value not in ("", "unset"):
|
|
return dt.datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(dt.timezone.utc)
|
|
return None
|
|
|
|
|
|
def now() -> dt.datetime:
|
|
return dt.datetime.now(dt.timezone.utc)
|
|
|
|
|
|
def sev_rank(sev: str) -> int:
|
|
return SEVERITIES.index(sev) if sev in SEVERITIES else len(SEVERITIES)
|