risk-register-validator.py

#!/usr/bin/env python3
"""Validate required fields and basic values in a cybersecurity risk-register CSV."""

import argparse
import csv
from datetime import date, datetime
from pathlib import Path

REQUIRED = {"risk_id", "description", "owner", "likelihood", "impact", "treatment", "status", "review_date"}
LEVELS = {"low", "medium", "high", "critical", "1", "2", "3", "4", "5"}
STATUSES = {"open", "accepted", "mitigating", "transferred", "closed"}


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", type=Path)
    parser.add_argument("--delimiter", default=",")
    parser.add_argument("--allow-past-review", action="store_true")
    return parser.parse_args()


def main():
    args = parse_args()
    if len(args.delimiter) != 1:
        raise SystemExit("--delimiter must be one character")
    failures = []
    identifiers = set()
    with args.file.open(newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle, delimiter=args.delimiter)
        if reader.fieldnames and len(reader.fieldnames) != len(set(reader.fieldnames)):
            print("ERROR duplicate column names in CSV header")
            return 2
        missing = REQUIRED - set(reader.fieldnames or [])
        if missing:
            print("ERROR missing columns: " + ", ".join(sorted(missing)))
            return 2
        for line, row in enumerate(reader, start=2):
            for field in REQUIRED:
                if not (row.get(field) or "").strip():
                    failures.append(f"line {line}: empty {field}")
            identifier = row["risk_id"].strip()
            if identifier in identifiers:
                failures.append(f"line {line}: duplicate risk_id {identifier}")
            identifiers.add(identifier)
            if row["likelihood"].strip().lower() not in LEVELS:
                failures.append(f"line {line}: invalid likelihood")
            if row["impact"].strip().lower() not in LEVELS:
                failures.append(f"line {line}: invalid impact")
            if row["status"].strip().lower() not in STATUSES:
                failures.append(f"line {line}: invalid status")
            try:
                review = datetime.strptime(row["review_date"].strip(), "%Y-%m-%d").date()
                if not args.allow_past_review and row["status"].strip().lower() != "closed" and review < date.today():
                    failures.append(f"line {line}: review_date is overdue")
            except ValueError:
                failures.append(f"line {line}: review_date must be YYYY-MM-DD")

    if failures:
        print("\n".join(f"FAIL {item}" for item in failures))
        print(f"Validation failed: {len(failures)} issue(s)")
        return 1
    print(f"OK: {len(identifiers)} risk(s) validated")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())