audit-labs/control-coverage

 1"""Load an evidence corpus from audit-report JSON reports.
 2
 3audit-report emits one JSON document per evidence package. Each finding carries
 4the controls it maps to and a status::
 5
 6    {"findings": [
 7        {"id": "github.org.require-2fa", "status": "pass", "severity": "high",
 8         "controls": ["SOC2:CC6.1", "ISO:A.5.17", "NIST:IA-2"], ...},
 9        ...
10    ]}
11
12A **corpus** is any number of these reports — typically one per platform and date
13(AWS, GitHub, GitLab…) — flattened into a list of :class:`Observation`, one per
14(finding, control) pair. Coverage is then computed by joining observations onto a
15framework catalog.
16"""
17
18from __future__ import annotations
19
20import json
21from dataclasses import dataclass
22from pathlib import Path
23
24# Statuses as emitted by audit-report's engine.
25PASS = "pass"
26FAIL = "fail"
27NOT_APPLICABLE = "not_applicable"
28
29
30@dataclass(frozen=True)
31class Observation:
32    """One finding's bearing on one control, with provenance."""
33
34    control: str  # "FRAMEWORK:ID"
35    status: str  # pass | fail | not_applicable
36    rule_id: str
37    title: str
38    severity: str
39    source: str  # report file / package name the finding came from
40
41
42def _iter_report(doc: dict, source: str):
43    for f in doc.get("findings", []):
44        status = f.get("status", NOT_APPLICABLE)
45        rule_id = f.get("id", "")
46        title = f.get("title", "")
47        severity = f.get("severity", "medium")
48        for control in f.get("controls", []):
49            yield Observation(
50                control=control,
51                status=status,
52                rule_id=rule_id,
53                title=title,
54                severity=severity,
55                source=source,
56            )
57
58
59def load_report(path: str | Path) -> list[Observation]:
60    """Load observations from a single audit-report JSON file."""
61    p = Path(path)
62    doc = json.loads(p.read_text(encoding="utf-8"))
63    # Prefer the package name audit-report records; fall back to the file name.
64    source = doc.get("source_package") or p.name
65    return list(_iter_report(doc, source))
66
67
68def _expand(paths: list[str | Path]) -> list[Path]:
69    """Resolve inputs: JSON files pass through; directories contribute their *.json."""
70    resolved: list[Path] = []
71    for raw in paths:
72        p = Path(raw)
73        if p.is_dir():
74            resolved.extend(sorted(p.glob("*.json")))
75        else:
76            resolved.append(p)
77    return resolved
78
79
80def load_corpus(paths: list[str | Path]) -> list[Observation]:
81    """Load and flatten observations from files and/or directories of reports."""
82    files = _expand(paths)
83    if not files:
84        raise ValueError("no audit-report JSON files found in the given paths")
85    observations: list[Observation] = []
86    for f in files:
87        observations.extend(load_report(f))
88    return observations