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>
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Has the inbox spoken more recently than the register has looked?
|
|
|
|
RISK-WP-0004-T02. On 2026-08-19 this register graded RISK-F-0001 `critical`
|
|
while its fix notice sat unread in the inbox. The residual said: if it slips
|
|
again, make it a check. It slipped once, so here is the check.
|
|
|
|
Mechanical only. It compares timestamps and matches on the system name; it does
|
|
not read the message and does not decide anything. It reports findings whose
|
|
inbox has spoken since the register last looked.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import register_lib as lib
|
|
|
|
HUB = "http://127.0.0.1:8000/messages/?to_agent=risk-nexus&limit=100"
|
|
|
|
|
|
def fetch() -> list[dict] | None:
|
|
try:
|
|
with urllib.request.urlopen(HUB, timeout=6) as r:
|
|
return json.load(r)
|
|
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
|
|
def main() -> int:
|
|
msgs = fetch()
|
|
if msgs is None:
|
|
# Fail loud: an unreachable inbox means no check can be certified current.
|
|
print("Inbox: UNREACHABLE — the hub did not answer.")
|
|
print(" No check can be certified current while this is true.")
|
|
print(" Findings may have moved without the register hearing about it.")
|
|
return 0
|
|
|
|
fs = [f for f in lib.findings() if lib.watched(f.get("status"))]
|
|
rows = []
|
|
for f in fs:
|
|
looked = lib.moment(f.get("last_checked"))
|
|
system = str(f.get("system", "")).lower()
|
|
owner = str(f.get("fix_owner", "")).lower()
|
|
if not looked or not system:
|
|
continue
|
|
for m in msgs:
|
|
said = lib.moment(m.get("created_at"))
|
|
if not said or said <= looked:
|
|
continue
|
|
blob = f"{m.get('subject','')} {m.get('body','')} {m.get('from_agent','')}".lower()
|
|
if system in blob or (owner and owner in blob) or f["id"].lower() in blob:
|
|
rows.append(
|
|
f"{f['id']} — last looked {looked:%Y-%m-%d %H:%MZ}; "
|
|
f"{m.get('from_agent')} wrote {said:%Y-%m-%d %H:%MZ}: {m.get('subject','')[:70]}"
|
|
)
|
|
break
|
|
|
|
print(f"Inbox: {len(msgs)} message(s) addressed to risk-nexus\n")
|
|
if rows:
|
|
print("Findings whose inbox has spoken since the register last looked:")
|
|
for r in rows:
|
|
print(f" {r}")
|
|
print("\n Read those before grading or checking. This is question zero.")
|
|
else:
|
|
print("No finding has an inbox message newer than its last check.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|