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, args, 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, FileNotFoundError, 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, FileNotFoundError, 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    formats = [f.strip() for f in args.format.split(",") if f.strip()]
163    if args.out:
164        out_dir = Path(args.out)
165        out_dir.mkdir(parents=True, exist_ok=True)
166        stem = _slug(subject) or "coverage"
167        for fmt in formats:
168            ext = reporters.EXTENSIONS.get(fmt, fmt)
169            name = "soa" if fmt == "soa" else "coverage"
170            path = out_dir / f"{name}.{ext}" if fmt == "soa" else out_dir / f"{stem}.{ext}"
171            path.write_text(reporters.render(report, fmt), encoding="utf-8")
172            print(f"wrote {path}")
173    else:
174        # Print the first requested format to stdout.
175        print(reporters.render(report, formats[0]), end="")
176
177    return _exit_code(report, args.fail_under)
178
179
180def _trend_mode(args, scp, names, subject, current) -> int:
181    from . import trend
182
183    baseline = _build_report([args.baseline], args, scp, names, subject)
184    tr = trend.compare(baseline, current)
185
186    formats = [f.strip() for f in args.format.split(",") if f.strip()]
187    renderers = {
188        "md": trend.render_markdown,
189        "html": trend.render_html,
190        "json": trend.render_json,
191    }
192    unknown = [f for f in formats if f not in renderers]
193    if unknown:
194        raise SystemExit(f"error: trend mode supports md, html, and json, not: {', '.join(unknown)}")
195
196    if args.out:
197        out_dir = Path(args.out)
198        out_dir.mkdir(parents=True, exist_ok=True)
199        for fmt in formats:
200            path = out_dir / f"trend.{fmt}"
201            path.write_text(renderers[fmt](tr), encoding="utf-8")
202            print(f"wrote {path}")
203    else:
204        print(renderers[formats[0]](tr), end="")
205
206    if args.fail_on_regression and tr.total_regressions:
207        print(
208            f"trend gate: {tr.total_regressions} control(s) regressed or lost coverage",
209            file=sys.stderr,
210        )
211        return 1
212    return 0
213
214
215def _crosswalk_mode(args, report) -> int:
216    from . import crosswalk
217
218    xw = crosswalk.build(report)
219    formats = [f.strip() for f in args.format.split(",") if f.strip()]
220    renderers = {
221        "md": crosswalk.render_markdown,
222        "html": crosswalk.render_html,
223        "json": crosswalk.render_json,
224    }
225    unknown = [f for f in formats if f not in renderers]
226    if unknown:
227        raise SystemExit(
228            f"error: crosswalk mode supports md, html, and json, not: {', '.join(unknown)}"
229        )
230
231    if args.out:
232        out_dir = Path(args.out)
233        out_dir.mkdir(parents=True, exist_ok=True)
234        for fmt in formats:
235            path = out_dir / f"crosswalk.{fmt}"
236            path.write_text(renderers[fmt](xw), encoding="utf-8")
237            print(f"wrote {path}")
238    else:
239        print(renderers[formats[0]](xw), end="")
240    return 0
241
242
243def _exit_code(report, fail_under: float | None) -> int:
244    if fail_under is None:
245        return 0
246    below = [fc for fc in report.frameworks if fc.coverage_pct < fail_under]
247    if below:
248        for fc in below:
249            print(
250                f"coverage gate: {fc.catalog.framework} at {fc.coverage_pct}% "
251                f"is below {fail_under}%",
252                file=sys.stderr,
253            )
254        return 1
255    return 0
256
257
258def _slug(text: str) -> str:
259    return "".join(c if c.isalnum() else "-" for c in text.lower()).strip("-")
260
261
262if __name__ == "__main__":
263    raise SystemExit(main())