#!/usr/bin/env python3 """Checks a lifecycle audit gap list before it is handed to a colleague. Usage: python3 scripts/gap_list_check.py GAPS.csv [--json] GAPS.csv uses the header of assets/gap-list-template.csv. The optional override_reason column holds the owner's written reason for moving a row out of score order, or the decisions.md exception that changes its priority. Python 3.9+, standard library only. Reads one file, calls no API, writes nothing. What it checks (rules: references/gap-list-and-change-requests.md): - required fields: id (unique), priority P0 to P3, rule IDs, FAIL or WARN, finding, evidence, owner, confidence, reach, effort and status values - score = reach_per_month x severity weight (FAIL 3, WARN 1) x confidence (confirmed 1.0, configuration-only 0.6, unconfirmed 0.3) / effort (S 1, M 2, L 4) recomputed for every row and compared with the score column - P0 and P1 rows name a change request (CR-) and a test - every FAIL sits at P0 or P1, and a FAIL on PC02, PC08, PC09, DL05, DL06, DL07, DL10 or DL16 sits at P0, unless override_reason records the exception - rows appear in priority order and, within a priority, in score order - cells that look like email addresses (customer contact details do not belong in a gap list; the value is never printed) ERROR lines must be fixed before hand-over; NOTE lines need a look. Exit status: 1 when any ERROR, 2 on bad input, 0 otherwise. """ import argparse import csv import json import math import re import sys REQUIRED = ["id", "priority", "area", "rule_id", "severity", "finding", "evidence", "confidence", "reach_per_month", "effort", "score", "owner", "proposed_change", "test", "status", "decision_needed"] PRIORITIES = ["P0", "P1", "P2", "P3"] WEIGHT = {"FAIL": 3, "WARN": 1} CONFIDENCE = {"confirmed": 1.0, "configuration-only": 0.6, "unconfirmed": 0.3} EFFORT = {"S": 1, "M": 2, "L": 4} STATUSES = {"open", "in progress", "verified", "accepted risk"} RULE = re.compile(r"^(JC|TD|SG|FC|PC|DL|MH)\d{2}$") P0_ON_FAIL = {"PC02", "PC08", "PC09", "DL05", "DL06", "DL07", "DL10", "DL16"} EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+\.[A-Za-z0-9.-]+") def bad_input(message): print("gap_list_check: " + message, file=sys.stderr) sys.exit(2) def number(text): value = float(text.replace(",", "").strip()) if not math.isfinite(value): # nan and inf parse as floats but are not counts raise ValueError(text) return value def shown(value): return format(round(value), ",") def read_rows(path): try: with open(path, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) header = [h.strip().lower() for h in (reader.fieldnames or [])] missing = [c for c in REQUIRED if c not in header] if missing: bad_input("missing columns: " + ", ".join(missing)) rows = [] for line, raw in enumerate(reader, start=2): row = {(k or "").strip().lower(): (v or "").strip() for k, v in raw.items() if k is not None} row["_line"] = line rows.append(row) except (OSError, UnicodeDecodeError, csv.Error) as exc: bad_input("cannot read %s as a UTF-8 CSV: %s (save Excel files as CSV UTF-8)" % (path, exc)) if not rows: bad_input("no data rows in %s" % path) return rows def check(rows): problems = [] def report(level, row, text): problems.append({"level": level, "id": row.get("id") or "line %d" % row["_line"], "detail": text}) seen = set() for row in rows: rid = row.get("id", "") if not rid: report("ERROR", row, "id is blank.") elif rid in seen: report("ERROR", row, "id is used by more than one row.") seen.add(rid) priority = row["priority"].upper() severity = row["severity"].upper() effort = row["effort"].upper() confidence = row["confidence"].lower() override = row.get("override_reason", "") if priority not in PRIORITIES: report("ERROR", row, "priority '%s' is not P0, P1, P2 or P3." % row["priority"]) if severity not in WEIGHT: report("ERROR", row, "severity '%s' is not FAIL or WARN (INFO and PASS stay out of the gap" " list)." % row["severity"]) rules = [r.strip().upper() for r in re.split(r"[;,]", row["rule_id"]) if r.strip()] if not rules: report("ERROR", row, "rule_id is blank.") for rule in rules: if not RULE.match(rule): report("ERROR", row, "rule_id '%s' is not a rule ID such as PC02." % rule) for column in ("finding", "evidence", "owner"): if not row[column]: report("ERROR", row, "%s is blank." % column) if confidence not in CONFIDENCE: report("ERROR", row, "confidence '%s' is not confirmed, configuration-only or unconfirmed." % row["confidence"]) if effort not in EFFORT: report("ERROR", row, "effort '%s' is not S, M or L." % row["effort"]) if row["status"] and row["status"].lower() not in STATUSES: report("ERROR", row, "status '%s' is not open, in progress, verified or accepted risk." % row["status"]) elif not row["status"]: report("NOTE", row, "status is blank; use open until someone picks it up.") try: reach = number(row["reach_per_month"]) if reach < 0: raise ValueError except ValueError: report("ERROR", row, "reach_per_month '%s' is not a number of 0 or more." % row["reach_per_month"]) reach = None row["_score"] = None if reach is not None and severity in WEIGHT and confidence in CONFIDENCE and effort in EFFORT: computed = reach * WEIGHT[severity] * CONFIDENCE[confidence] / EFFORT[effort] row["_score"] = computed if not row["score"]: report("NOTE", row, "score is blank; the formula gives %s." % shown(computed)) else: try: given = number(row["score"]) if abs(given - computed) > max(1.0, 0.01 * computed): report("ERROR", row, "score column says %s; reach x weight x confidence / effort gives %s." % (row["score"], shown(computed))) except ValueError: report("ERROR", row, "score '%s' is not a number." % row["score"]) if priority in ("P0", "P1"): if "CR-" not in row["proposed_change"].upper(): report("ERROR", row, "%s rows need a change request: name its CR ID in proposed_change." % priority) if not row["test"]: report("ERROR", row, "%s rows need the test that proves the fix." % priority) if severity == "FAIL" and priority in ("P2", "P3") and not override: report("ERROR", row, "a FAIL belongs at P0 or P1; move it, or record the documented exception in" " override_reason.") p0_rules = sorted(set(rules) & P0_ON_FAIL) if severity == "FAIL" and p0_rules and priority != "P0" and not override: report("ERROR", row, "a FAIL on %s is P0 under the audit's rules (sending without permission, a" " rejection-level authentication failure or provider rejection on a bulk stream, or a spam" " rate of 0.30%% or more); move it, or record the exception in override_reason." % ", ".join(p0_rules)) if severity == "WARN" and priority == "P0": report("NOTE", row, "a WARN at P0: confirm it is wrong-recipient or wrong-account content, or" " lower it.") for column, value in row.items(): if not column.startswith("_") and EMAIL.search(value or ""): report("NOTE", row, "%s contains something that looks like an email address (not printed);" " keep only sender addresses and internal IDs in a gap list." % column) break rank = {p: i for i, p in enumerate(PRIORITIES)} previous = None for row in rows: priority = row["priority"].upper() if priority not in rank: continue if previous is not None: prev_priority = previous["priority"].upper() if rank[priority] < rank[prev_priority]: report("NOTE", row, "%s row follows a %s row; list rows from P0 to P3." % (priority, prev_priority)) elif (priority == prev_priority and row["_score"] is not None and previous["_score"] is not None and row["_score"] > previous["_score"] + 0.5 and not row.get("override_reason")): report("NOTE", row, "score %s is above the row before it (%s, %s); move it up or record the" " owner's reason in override_reason." % ( shown(row["_score"]), previous["id"], shown(previous["_score"]))) previous = row return problems def main(): parser = argparse.ArgumentParser(description="Check a lifecycle audit gap list CSV.") parser.add_argument("gaps") parser.add_argument("--json", action="store_true") args = parser.parse_args() rows = read_rows(args.gaps) problems = check(rows) rank = {p: i for i, p in enumerate(PRIORITIES)} ordered = sorted((r for r in rows if r["priority"].upper() in rank), key=lambda r: (rank[r["priority"].upper()], -(r["_score"] or 0))) counts = {p: sum(1 for r in rows if r["priority"].upper() == p) for p in PRIORITIES} errors = sum(1 for p in problems if p["level"] == "ERROR") notes = len(problems) - errors summary = "%d rows: %s; %d ERROR, %d NOTE. Nothing was changed." % ( len(rows), ", ".join("%d %s" % (counts[p], p) for p in PRIORITIES), errors, notes) if args.json: print(json.dumps({ "file": args.gaps, "problems": problems, "counts": counts, "summary": summary, "order": [{"id": r["id"], "priority": r["priority"].upper(), "score": round(r["_score"]) if r["_score"] is not None else None} for r in ordered], }, indent=2)) else: print("Gap list check: %s" % args.gaps) print("") for p in sorted(problems, key=lambda p: (p["level"] != "ERROR", p["id"])): print("%s %s: %s" % (p["level"], p["id"], p["detail"])) if problems: print("") print("Order by priority, then score:") for r in ordered: print(" %s %s score %s %s %s: %s (owner: %s)" % ( r["priority"].upper(), r["id"], shown(r["_score"]) if r["_score"] is not None else "?", r["rule_id"], r["severity"].upper(), r["finding"], r["owner"] or "none")) print("") print(summary) sys.exit(1 if errors else 0) if __name__ == "__main__": main()