audit-labs/control-coverage

  1"""Command-line entry point for control-coverage."""
  2
  3from __future__ import annotations
  4
  5import argparse
  6import sys
  7from datetime import datetime, timezone
  8from pathlib import Path
  9
 10from . import __version__, catalog, corpus, reporters, scope
 11from .coverage import evaluate
 12
 13
 14def _parse_args(argv: list[str]) -> argparse.Namespace:
 15    parser = argparse.ArgumentParser(
 16        prog="control-coverage",
 17        description=(
 18            "Score an evidence corpus against complete framework catalogs: what "
 19            "share of each framework does the evidence address, and which controls "
 20            "are blind spots no finding touches?"
 21        ),
 22    )
 23    parser.add_argument(
 24        "reports",
 25        nargs="+",
 26        help="audit-report JSON files, and/or directories containing them",
 27    )
 28    parser.add_argument(
 29        "--framework",
 30        help=(
 31            "comma-separated frameworks to evaluate (e.g. SOC2,ISO,NIST). "
 32            "Default: every framework the corpus cites, or the scope file's list."
 33        ),
 34    )
 35    parser.add_argument(
 36        "--scope",
 37        help="path to a scope / Statement of Applicability YAML (marks exclusions)",
 38    )
 39    parser.add_argument(
 40        "--subject",
 41        help="name for the subject of this corpus (overrides the scope file)",
 42    )
 43    parser.add_argument(
 44        "--format",
 45        default="md",
 46        help="comma-separated output formats: md, html, json, soa (default: md)",
 47    )
 48    parser.add_argument(
 49        "--out",
 50        help="directory to write reports into (default: print the first format to stdout)",
 51    )
 52    parser.add_argument(
 53        "--blind-spots",
 54        action="store_true",
 55        help="print only the unaddressed in-scope controls, then exit",
 56    )
 57    parser.add_argument(
 58        "--baseline",
 59        metavar="PATH",
 60        help=(
 61            "trend mode: an earlier corpus (file or directory) to compare against. "
 62            "Reports how coverage moved — what improved, regressed, was gained or lost."
 63        ),
 64    )
 65    parser.add_argument(
 66        "--crosswalk",
 67        action="store_true",
 68        help=(
 69            "crosswalk mode: show which controls each piece of evidence supports "
 70            "across frameworks, and the minimal evidence set that covers them all"
 71        ),
 72    )
 73    parser.add_argument(
 74        "--fail-under",
 75        type=float,
 76        metavar="PCT",
 77        help="exit non-zero if any framework's coverage %% is below PCT (CI gate)",
 78    )
 79    parser.add_argument(
 80        "--fail-on-regression",
 81        action="store_true",
 82        help="in trend mode, exit non-zero if any control regressed or lost coverage",
 83    )
 84    parser.add_argument("--version", action="version", version=f"control-coverage {__version__}")
 85    return parser.parse_args(argv)
 86
 87
 88def _select_frameworks(args, observations, scp) -> list[str]:
 89    """Decide which framework catalogs to load, in priority order."""
 90    if args.framework:
 91        return [f.strip() for f in args.framework.split(",") if f.strip()]
 92    if scp.frameworks:
 93        return scp.frameworks
 94    # Infer from the corpus: every framework prefix the observations cite.
 95    cited = sorted({o.control.split(":", 1)[0] for o in observations if ":" in o.control})
 96    if not cited:
 97        raise SystemExit(
 98            "error: could not infer frameworks from the corpus. Pass --framework."
 99        )
100    return cited
101
102
103def _print_blind_spots(report) -> None:
104    total = 0
105    for fc in report.frameworks:
106        spots = fc.blind_spots
107        if not spots:
108            continue
109        print(f"{fc.catalog.name} — {len(spots)} unaddressed:")
110        for r in spots:
111            print(f"  {r.control.code}  {r.control.title}")
112        total += len(spots)
113    print(f"\n{total} in-scope control(s) unaddressed across {len(report.frameworks)} framework(s).")
114
115
116def _now() -> str:
117    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
118
119
120def _build_report(paths, scp, names, subject):
121    """Load a corpus from *paths* and evaluate it into a CoverageReport."""
122    try:
123        observations = corpus.load_corpus(paths)
124    except (ValueError, OSError) as exc:
125        raise SystemExit(f"error: {exc}") from None
126    catalogs = catalog.load_frameworks(names)
127    return evaluate(catalogs, observations, scope=scp, subject=subject, generated_at=_now())
128
129
130def main(argv: list[str] | None = None) -> int:
131    args = _parse_args(sys.argv[1:] if argv is None else argv)
132
133    if args.crosswalk and args.baseline:
134        raise SystemExit("error: --crosswalk and --baseline cannot be combined")
135
136    try:
137        observations = corpus.load_corpus(args.reports)
138    except (ValueError, OSError) as exc:
139        raise SystemExit(f"error: {exc}") from None
140
141    scp = scope.load(args.scope) if args.scope else scope.empty()
142
143    try:
144        names = _select_frameworks(args, observations, scp)
145        catalogs = catalog.load_frameworks(names)
146    except ValueError as exc:
147        raise SystemExit(f"error: {exc}") from None
148
149    subject = args.subject or scp.subject
150    report = evaluate(catalogs, observations, scope=scp, subject=subject, generated_at=_now())
151
152    if args.baseline:
153        return _trend_mode(args, scp, names, subject, report)
154
155    if args.crosswalk:
156        return _crosswalk_mode(args, report)
157
158    if args.blind_spots:
159        _print_blind_spots(report)
160        return _exit_code(report, args.fail_under)
161
162    return _default_mode(args, report, subject)
163
164
165def _default_mode(args, report, subject) -> int:
166    formats = [f.strip() for f in args.format.split(",") if f.strip()]
167    if args.out:
168        out_dir = Path(args.out)
169        out_dir.mkdir(parents=True, exist_ok=True)
170        stem = _slug(subject) or "coverage"
171        for fmt in formats:
172            ext = reporters.EXTENSIONS.get(fmt, fmt)
173            name = "soa" if fmt == "soa" else "coverage"
174            path = out_dir / f"{name}.{ext}" if fmt == "soa" else out_dir / f"{stem}.{ext}"
175            path.write_text(reporters.render(report, fmt), encoding="utf-8")
176            print(f"wrote {path}")
177    else:
178        # Print the first requested format to stdout.
179        print(reporters.render(report, formats[0]), end="")
180
181    return _exit_code(report, args.fail_under)
182
183
184def _trend_mode(args, scp, names, subject, current) -> int:
185    from . import trend
186
187    baseline = _build_report([args.baseline], scp, names, subject)
188    tr = trend.compare(baseline, current)
189
190    formats = [f.strip() for f in args.format.split(",") if f.strip()]
191    renderers = {
192        "md": trend.render_markdown,
193        "html": trend.render_html,
194        "json": trend.render_json,
195    }
196    unknown = [f for f in formats if f not in renderers]
197    if unknown:
198        raise SystemExit(f"error: trend mode supports md, html, and json, not: {', '.join(unknown)}")
199
200    if args.out:
201        out_dir = Path(args.out)
202        out_dir.mkdir(parents=True, exist_ok=True)
203        for fmt in formats:
204            path = out_dir / f"trend.{fmt}"
205            path.write_text(renderers[fmt](tr), encoding="utf-8")
206            print(f"wrote {path}")
207    else:
208        print(renderers[formats[0]](tr), end="")
209
210    if args.fail_on_regression and tr.total_regressions:
211        print(
212            f"trend gate: {tr.total_regressions} control(s) regressed or lost coverage",
213            file=sys.stderr,
214        )
215        return 1
216    return 0
217
218
219def _crosswalk_mode(args, report) -> int:
220    from . import crosswalk
221
222    xw = crosswalk.build(report)
223    formats = [f.strip() for f in args.format.split(",") if f.strip()]
224    renderers = {
225        "md": crosswalk.render_markdown,
226        "html": crosswalk.render_html,
227        "json": crosswalk.render_json,
228    }
229    unknown = [f for f in formats if f not in renderers]
230    if unknown:
231        raise SystemExit(
232            f"error: crosswalk mode supports md, html, and json, not: {', '.join(unknown)}"
233        )
234
235    if args.out:
236        out_dir = Path(args.out)
237        out_dir.mkdir(parents=True, exist_ok=True)
238        for fmt in formats:
239            path = out_dir / f"crosswalk.{fmt}"
240            path.write_text(renderers[fmt](xw), encoding="utf-8")
241            print(f"wrote {path}")
242    else:
243        print(renderers[formats[0]](xw), end="")
244    return 0
245
246
247def _exit_code(report, fail_under: float | None) -> int:
248    if fail_under is None:
249        return 0
250    below = [fc for fc in report.frameworks if fc.coverage_pct < fail_under]
251    if below:
252        for fc in below:
253            print(
254                f"coverage gate: {fc.catalog.framework} at {fc.coverage_pct}% "
255                f"is below {fail_under}%",
256                file=sys.stderr,
257            )
258        return 1
259    return 0
260
261
262def _slug(text: str) -> str:
263    return "".join(c if c.isalnum() else "-" for c in text.lower()).strip("-")
264
265
266if __name__ == "__main__":
267    raise SystemExit(main())