diff --git a/Makefile b/Makefile index 15dcb55..fe33544 100644 --- a/Makefile +++ b/Makefile @@ -1,35 +1,72 @@ PY := python3 TOOLS := tools -.PHONY: help register check checked due fixes coverage test +# Parameters for the query and write targets (see `make help` for examples). +ID ?= +SEV ?= +SYSTEM ?= +OUTCOME ?= +NOTE ?= -help: - @echo "make register - rebuild REGISTER.md from findings/" - @echo "make check - verify the index, report what is going quiet, and read the inbox" - @echo "make due - just the work list: what needs a check right now" - @echo "make fixes - the state of every fix this register claims to track" - @echo "make coverage - what the register has never heard from" - @echo "make checked - record a check outcome: make checked ARGS=\"RISK-F-0002 clean\"" +.DEFAULT_GOAL := help +.PHONY: help list all show waits escalations policies check due fixes coverage \ + register checked test -register: - @$(PY) $(TOOLS)/register_index.py +##@ Inspect the register (read-only) -check: +list: ## Live findings, most severe first. Filters: make list SEV=high | make list SYSTEM=qonto-assistant + @$(PY) $(TOOLS)/risk.py list $(if $(SEV),--severity $(SEV)) $(if $(SYSTEM),--system $(SYSTEM)) + +all: ## Every finding, including fixed and withdrawn + @$(PY) $(TOOLS)/risk.py list --all $(if $(SEV),--severity $(SEV)) $(if $(SYSTEM),--system $(SYSTEM)) + +show: ## One finding or regulatory record: make show ID=RISK-F-0012 + @test -n "$(ID)" || { echo 'usage: make show ID=RISK-F-0012'; exit 2; } + @$(PY) $(TOOLS)/risk.py show $(ID) + +waits: ## What the register waits on, or owes, ordered by default date + @$(PY) $(TOOLS)/risk.py waits + +escalations: ## Escalations and their delivery state + @$(PY) $(TOOLS)/risk.py escalations + +policies: ## Regulatory records and full legal policies with next review + @$(PY) $(TOOLS)/risk.py policies + +##@ Run the register + +check: ## Full sweep: verify the index, report due and quiet work, read the inbox @$(PY) $(TOOLS)/check_all.py -checked: - @$(PY) $(TOOLS)/record_check.py $(ARGS) - @$(PY) $(TOOLS)/register_index.py - -due: +due: ## Just the work list: what needs a check right now @$(PY) $(TOOLS)/register_check.py @$(PY) $(TOOLS)/inbox_check.py -fixes: +fixes: ## State of every fix the register tracks, read from owner workplans @$(PY) $(TOOLS)/fix_tracker.py -coverage: +coverage: ## Which estate repos the register has never heard from @$(PY) $(TOOLS)/coverage.py -test: +##@ Record and maintain + +checked: ## Record a check: make checked ID=RISK-F-0012 OUTCOME=clean|moved NOTE="what was seen" + @if [ -n "$(ARGS)" ]; then $(PY) $(TOOLS)/record_check.py $(ARGS); \ + else test -n "$(ID)" -a -n "$(OUTCOME)" || { echo 'usage: make checked ID=RISK-F-0012 OUTCOME=clean NOTE="..."'; exit 2; }; \ + $(PY) $(TOOLS)/record_check.py $(ID) $(OUTCOME) $(if $(NOTE),"$(NOTE)"); fi + @$(PY) $(TOOLS)/register_index.py + +register: ## Rebuild REGISTER.md from findings/ + @$(PY) $(TOOLS)/register_index.py + +test: ## Run the tool test suite @PYTHONDONTWRITEBYTECODE=1 $(PY) -m unittest discover -s tests -v + +##@ Help + +help: ## List targets with examples (the default) + @awk 'BEGIN {FS = ":.*## "} \ + /^##@/ {printf "\n\033[1m%s\033[0m\n", substr($$0, 5); next} \ + /^[a-z][a-zA-Z_-]*:.*## / {printf " \033[36m%-12s\033[0m %s\n", $$1, $$2}' $(MAKEFILE_LIST) + @echo + @echo "Defer a check (operator decision): make checked ARGS='RISK-F-0012 defer 2026-10-15 \"reason\"'" diff --git a/README.md b/README.md index 28b695c..b0dadd1 100644 --- a/README.md +++ b/README.md @@ -32,10 +32,23 @@ does not host: `policy-nexus` is the publication surface. ## Using it ``` -make register # rebuild REGISTER.md from findings/ -make check # verify the index, then report what is going quiet +make # list every target with an example +make list # live findings, most severe first +make list SEV=high # filter by severity (or SYSTEM=qonto-assistant) +make all # include fixed and withdrawn +make show ID=RISK-F-0012 # one finding or regulatory record +make waits # who owes what, by default date +make escalations # escalations and their delivery state +make policies # regulatory records and next reviews +make check # full sweep: index, due work, inbox +make checked ID=RISK-F-0012 OUTCOME=clean NOTE="what was seen" +make register # rebuild REGISTER.md from findings/ ``` +The query targets use `tools/risk.py`, which is read-only. Its verbs (`list`, +`show`) are generic so that a future common nexus interface can map onto them. +Writes go only through `make checked`. + `make check` reports ungraded findings, overdue reviews, stalled remediation, embargoes due for re-decision, escalations awaiting the operator, and what is owed at the production transition. It changes nothing. diff --git a/tests/test_risk_cli.py b/tests/test_risk_cli.py new file mode 100644 index 0000000..50bdbe8 --- /dev/null +++ b/tests/test_risk_cli.py @@ -0,0 +1,44 @@ +import contextlib +import io +import pathlib +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "tools")) +import register_lib as lib +import risk + +LIVE = {"id": "RISK-F-0002", "status": "open", "severity": "medium", "fix_owner": "b", + "title": "live", "_path": lib.REPO / "findings" / "x.md", + "waiting_on": [{"who": "b", "what": "answer", "default_at": "2026-10-06"}]} +FIXED = {"id": "RISK-F-0001", "status": "fixed", "severity": "high", "fix_owner": "a", + "title": "done", "_path": lib.REPO / "findings" / "y.md"} + + +class RiskCliTests(unittest.TestCase): + def run_cli(self, *argv): + out = io.StringIO() + with patch.object(lib, "findings", return_value=[LIVE, FIXED]), \ + patch.object(lib, "regulatory", return_value=[]), \ + patch.object(lib, "workplans", return_value=[]), \ + contextlib.redirect_stdout(out): + code = risk.main(list(argv)) + return code, out.getvalue() + + def test_list_hides_closed_unless_all(self): + self.assertNotIn("RISK-F-0001", self.run_cli("list")[1]) + self.assertIn("RISK-F-0001", self.run_cli("list", "--all", "--severity", "high")[1]) + + def test_show_known_and_unknown(self): + code, out = self.run_cli("show", "RISK-F-0002") + self.assertEqual(code, 0) + self.assertIn("waiting on b until 2026-10-06", out) + self.assertEqual(self.run_cli("show", "RISK-F-9999")[0], 1) + + def test_waits_ordered_by_deadline(self): + self.assertIn("2026-10-06 RISK-F-0002", self.run_cli("waits")[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/risk.py b/tools/risk.py new file mode 100644 index 0000000..491cfca --- /dev/null +++ b/tools/risk.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Read-only query CLI over the register: list, show, waits, escalations, policies. + + python3 tools/risk.py list [--all] [--severity high] [--system qonto-assistant] + python3 tools/risk.py show RISK-F-0012 + python3 tools/risk.py waits + python3 tools/risk.py escalations + python3 tools/risk.py policies + +The verbs are generic on purpose (list/show over records), so that a future +common nexus interface can map onto them. Writing stays with record_check.py. +""" +from __future__ import annotations + +import argparse +import sys + +import register_lib as lib + +SHOW_FIELDS = ("title", "status", "severity", "impact", "likelihood", "disclosure", + "system", "fix_owner", "fix_tracking", "closure_condition", "escalation", + "escalation_status", "cadence", "next_check", "ruling") + + +def cmd_list(a) -> int: + fs = lib.findings() + if not a.all: + fs = [f for f in fs if lib.watched(f.get("status"))] + if a.severity: + fs = [f for f in fs if f.get("severity") == a.severity] + if a.system: + fs = [f for f in fs if f.get("system") == a.system] + fs.sort(key=lambda f: (lib.sev_rank(f.get("severity")), f["id"])) + if not fs: + print("no matching findings") + for f in fs: + print(f"{f['id']} {str(f.get('severity')):8} {str(f.get('status')):9} " + f"{str(f.get('fix_owner')):20} {f.get('title')}") + return 0 + + +def cmd_show(a) -> int: + rec = next((r for r in lib.findings() + lib.regulatory() if r.get("id") == a.id), None) + if rec is None: + print(f"{a.id}: not found") + return 1 + print(f"{a.id} ({rec['_path'].relative_to(lib.REPO)})") + for key in SHOW_FIELDS: + if rec.get(key) not in (None, ""): + print(f" {key:18} {rec[key]}") + for w in rec.get("waiting_on") or []: + print(f" waiting on {w.get('who')} until {w.get('default_at')}: {w.get('what')}") + return 0 + + +def cmd_waits(_a) -> int: + rows = [] + for r in lib.records_with_waits(): + for w in r["waiting_on"]: + rows.append((str(w.get("default_at")), r.get("id"), w.get("who"), w.get("what"))) + for when, rid, who, what in sorted(rows): + print(f"{when} {rid:14} {who:20} {what}") + if not rows: + print("nobody is being waited on") + return 0 + + +def cmd_escalations(_a) -> int: + rows = [f for f in lib.findings() if f.get("escalation") not in (None, "none")] + for f in rows: + print(f"{f['id']} trigger {f.get('escalation_trigger')} {f.get('escalation_status')}" + f" sent {f.get('escalation_sent', '-')} {f.get('title')}") + if not rows: + print("no escalations recorded") + return 0 + + +def cmd_policies(_a) -> int: + for r in lib.regulatory(): + print(f"{r.get('id'):14} {str(r.get('status')):22} next {str(r.get('next_check'))[:16]:16} " + f"{r.get('title')}") + return 0 + + +def main(argv: list[str]) -> int: + p = argparse.ArgumentParser(prog="risk", description=__doc__.splitlines()[0]) + sub = p.add_subparsers(dest="cmd", required=True) + ls = sub.add_parser("list", help="findings, most severe first (live ones unless --all)") + ls.add_argument("--all", action="store_true", help="include fixed and withdrawn") + ls.add_argument("--severity", choices=lib.SEVERITIES) + ls.add_argument("--system") + ls.set_defaults(fn=cmd_list) + sh = sub.add_parser("show", help="one finding or regulatory record") + sh.add_argument("id") + sh.set_defaults(fn=cmd_show) + for name, fn, hlp in (("waits", cmd_waits, "everything owed to or by the register, by deadline"), + ("escalations", cmd_escalations, "escalations and their delivery state"), + ("policies", cmd_policies, "regulatory records and full policies")): + sub.add_parser(name, help=hlp).set_defaults(fn=fn) + a = p.parse_args(argv) + return a.fn(a) + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:]))