tegwick
a05ca6822b
T01 fix tracking now reads the owner's workplan file and found two findings the register should have known about. T02 incident and external report intake, the latter routed since the address is not ours to create. T03 the production transition defined by what is held rather than what was announced. T04 the README stops claiming a surface. T05 escalation carries a delivery state and is raised once when unacknowledged. T06 checked_by and a heartbeat, so a 1q rung cannot silently mean nobody looked. T07 coverage: 7 of 117 repos have ever appeared in a finding. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
211 lines
9.5 KiB
Python
211 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Report what the register is not saying out loud.
|
|
|
|
RISK-WP-0001-T08. Report only: it writes nothing, changes no field,
|
|
escalates nothing on its own, and blocks nothing. A human or the custodian
|
|
acts on the output. Exit code is 0 unless a finding cannot be read.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
import fix_tracker
|
|
import register_lib as lib
|
|
|
|
NOW = lib.now()
|
|
|
|
|
|
def main() -> int:
|
|
fs = [f for f in lib.findings() if lib.watched(f.get("status"))]
|
|
lines: list[str] = []
|
|
|
|
def section(title: str, rows: list[str], quiet: str) -> None:
|
|
lines.append(f"{title}:")
|
|
if rows:
|
|
lines.extend(f" {r}" for r in rows)
|
|
else:
|
|
lines.append(f" {quiet}")
|
|
lines.append("")
|
|
|
|
seen: dict[str, str] = {}
|
|
dupes = []
|
|
for f in lib.findings():
|
|
prior = seen.get(f["id"])
|
|
if prior:
|
|
dupes.append(f"{f['id']} — filed twice: {prior} and {f['_path'].name}; renumber the later commit")
|
|
seen[f["id"]] = f["_path"].name
|
|
if dupes:
|
|
section("Duplicate ids", dupes, "none")
|
|
|
|
unknown = [
|
|
f"{f['id']} — status '{f.get('status')}' is not one of {', '.join(lib.KNOWN_STATUSES)}; watched anyway"
|
|
for f in fs
|
|
if f.get("status") not in lib.KNOWN_STATUSES
|
|
]
|
|
if unknown:
|
|
section("Unrecognised status", unknown, "none")
|
|
|
|
# T03 — the `mitigated` defect was one instance of a class: the tooling
|
|
# quietly tolerating something it did not expect. Everything below is
|
|
# reported by name, and never silently ignored.
|
|
ids = {f["id"] for f in lib.findings()}
|
|
malformed: list[str] = []
|
|
for f in fs:
|
|
fid = f["id"]
|
|
for field in ("last_checked", "next_check", "embargo_since", "deferred_to"):
|
|
if f.get(field) and lib.moment(f[field]) is None:
|
|
malformed.append(f"{fid} — {field} is not a date: {f[field]!r}")
|
|
if (c := f.get("cadence")) and c not in lib.CADENCE_NAMES:
|
|
malformed.append(f"{fid} — cadence '{c}' is not a rung on the ladder")
|
|
if (d := f.get("disclosure")) and d not in ("public", "embargoed", "restricted", "unset"):
|
|
malformed.append(f"{fid} — disclosure '{d}' is not a defined state")
|
|
if (s := f.get("severity")) and s not in lib.SEVERITIES + ["unset"]:
|
|
malformed.append(f"{fid} — severity '{s}' is not on the scale")
|
|
if (on := f.get("constraint_on")) and on not in ids:
|
|
malformed.append(f"{fid} — constraint_on points at {on}, which does not exist")
|
|
for ref in (f.get("related") or []):
|
|
if ref not in ids:
|
|
malformed.append(f"{fid} — related names {ref}, which does not exist")
|
|
if f.get("embargo_condition") and f.get("disclosure") != "embargoed": # a lift is recorded as embargo_lifted
|
|
malformed.append(f"{fid} — carries an embargo_condition but disclosure is '{f.get('disclosure')}'")
|
|
if f.get("disclosure") == "embargoed" and not f.get("embargo_condition"):
|
|
malformed.append(f"{fid} — embargoed with no condition; a hold with no lift is a silence")
|
|
if f.get("escalation") == "required" and not f.get("escalation_trigger"):
|
|
malformed.append(f"{fid} — escalated with no trigger named")
|
|
if malformed:
|
|
section("Malformed", malformed, "none")
|
|
|
|
ungraded = [
|
|
f"{f['id']} — {', '.join(k for k in lib.GRADED_FIELDS if str(f.get(k, 'unset')) == 'unset')}"
|
|
for f in fs
|
|
if any(str(f.get(k, "unset")) == "unset" for k in lib.GRADED_FIELDS)
|
|
]
|
|
section("Ungraded", ungraded, "none — every watched finding carries a grade")
|
|
|
|
due, deferred = [], []
|
|
for f in fs:
|
|
if f.get("deferred_to"):
|
|
until = lib.moment(f["deferred_to"])
|
|
if until and NOW < until:
|
|
deferred.append(f"{f['id']} — deferred by the operator until {f['deferred_to']}")
|
|
continue
|
|
when = lib.moment(f.get("next_check"))
|
|
if when is None:
|
|
due.append(f"{f['id']} — no next_check set")
|
|
elif NOW >= when:
|
|
late = NOW - when
|
|
hours = int(late.total_seconds() // 3600)
|
|
rung = f.get("cadence", "instant")
|
|
due.append(
|
|
f"{f['id']} ({f.get('severity')}) — due {f['next_check']}"
|
|
f"{f', {hours}h late' if hours else ''}, cadence {rung}"
|
|
)
|
|
section("Checks due", due, "none")
|
|
if deferred:
|
|
section("Deferred by the operator", deferred, "none")
|
|
|
|
# The rung is the signal: how long this matter has held still.
|
|
stability = []
|
|
for f in sorted(fs, key=lambda f: lib.rung_index(f.get("cadence", "instant"))):
|
|
rung = f.get("cadence", "instant")
|
|
streak = f.get("clean_streak", 0)
|
|
note = " — at the ceiling" if rung == lib.TOP_RUNG else ""
|
|
stability.append(f"{f['id']}: {rung} ({streak} clean in a row){note}")
|
|
section("Stability — where each finding sits on the ladder", stability, "nothing watched")
|
|
|
|
bottom = [
|
|
f"{f['id']} ({f.get('severity')}) — still at the bottom rung since {f.get('last_checked')}"
|
|
for f in fs
|
|
if lib.rung_index(f.get("cadence", "instant")) == 0
|
|
and (m := lib.moment(f.get("last_checked")))
|
|
and NOW - m > dt.timedelta(days=14)
|
|
]
|
|
section("Stalled — escalation trigger 5", bottom, "none")
|
|
|
|
# Waits: typed, dated, and defaulted (docs/method/dependencies.md).
|
|
waiting, due_defaults, deep = [], [], []
|
|
# A closed finding can still owe something — RISK-F-0001 is `fixed` and
|
|
# still waiting on its publication entry. Waits outlive statuses, and they
|
|
# are not a findings-only idea: adopting a rule is nobody's finding.
|
|
with_waits = lib.records_with_waits()
|
|
waiting_ids = {f["id"] for f in with_waits}
|
|
for f in with_waits:
|
|
for w in (f.get("waiting_on") or []):
|
|
since = lib.moment(w.get("since"))
|
|
age = f"{(NOW - since).days}d" if since else "?"
|
|
when = lib.moment(w.get("default_at"))
|
|
waiting.append(f"{f['id']} — {w.get('who')}: {str(w.get('what'))[:64]} ({age} old, defaults {w.get('default_at')})")
|
|
if when and NOW >= when:
|
|
due_defaults.append(
|
|
f"{f['id']} — {w.get('who')} did not answer by {w.get('default_at')}; apply: {w.get('default')}"
|
|
)
|
|
# depth-two check: a record waiting on a record that is itself waiting
|
|
for ref in ([f.get("constraint_on")] if f.get("constraint_on") else []):
|
|
if ref in waiting_ids and f.get("waiting_on"):
|
|
deep.append(f"{f['id']} waits, and points at {ref} which also waits — depth two, cut one")
|
|
section("Waiting on someone", waiting, "nothing is waiting on anyone")
|
|
# RISK-WP-0005-T01: the fix's own state, read from the owner's workplan
|
|
# file rather than from our memory of what they told us.
|
|
section("Fix state", fix_tracker.report(), "no finding claims a tracked fix")
|
|
if due_defaults:
|
|
section("Defaults now due — apply them", due_defaults, "none")
|
|
if deep:
|
|
section("Dependency depth violations", deep, "none")
|
|
|
|
embargo = [
|
|
f"{f['id']} — lifts when: {f.get('embargo_condition')}"
|
|
for f in fs
|
|
if f.get("disclosure") == "embargoed"
|
|
]
|
|
section("Embargoed", embargo, "none")
|
|
|
|
# T05 — an escalation nobody acknowledged is indistinguishable from one
|
|
# never sent, which is the failure this register fixed for its own inbox
|
|
# and not, until now, for the path that matters more.
|
|
esc = []
|
|
for f in fs:
|
|
if f.get("escalation") != "required" or f.get("escalation_status") == "answered":
|
|
continue
|
|
sent = lib.moment(f.get("escalation_sent"))
|
|
age = f", sent {(NOW - sent).days}d ago" if sent else ", never marked sent"
|
|
state = f.get("escalation_status", "unknown")
|
|
tail = " ← unacknowledged; raise once more, then record the default" if (
|
|
sent and (NOW - sent).days >= 7 and state not in ("seen", "answered")
|
|
) else ""
|
|
esc.append(f"{f['id']} — trigger {f.get('escalation_trigger')}, {state}{age}{tail}")
|
|
section("Escalations awaiting the operator", esc, "none")
|
|
|
|
rescore = [
|
|
f"{f['id']} — {f.get('severity')} now, {f.get('severity_at_production')} at production"
|
|
for f in fs
|
|
if f.get("production_rescore")
|
|
]
|
|
section("Owed at the production transition", rescore, "none — no finding is graded lower for build mode")
|
|
|
|
reg = []
|
|
for r in lib.regulatory():
|
|
when = lib.moment(r.get("next_check")) or lib.moment(r.get("review_by"))
|
|
if when is None:
|
|
reg.append(f"{r.get('id')} — no next_check set")
|
|
elif NOW >= when:
|
|
reg.append(f"{r.get('id')} — due {when:%Y-%m-%d}: {str(r.get('title','')).strip()}")
|
|
section("Regulatory records due", reg, "none")
|
|
|
|
# T06 — a 1q rung means "stable for a quarter" and "nobody looked for a
|
|
# quarter", and those read identically. The heartbeat separates them.
|
|
checks = [m for f in lib.findings() if (m := lib.moment(f.get("last_checked")))]
|
|
if checks:
|
|
newest = max(checks)
|
|
quiet = (NOW - newest).days
|
|
if quiet >= 2:
|
|
lines.insert(0, "")
|
|
lines.insert(0, f" Nothing anywhere in the register has been checked for {quiet} days.")
|
|
lines.insert(0, "HEARTBEAT — THE LADDER IS NOT BEING CLIMBED:")
|
|
|
|
print(f"Register check — {NOW:%Y-%m-%d %H:%MZ}\n{len(fs)} live finding(s)\n")
|
|
print("\n".join(lines).rstrip())
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|