audit-labs/audit-report
Turn audit-tools evidence packages into control-mapped, auditor-ready reports.
clone: git clone https://gitbay.org/audit-labs/audit-report.git
v1.0.0: audit_report/diff.py · raw
1"""Diff mode — compare two evidence packages and report drift.
2
3Both packages are evaluated with the same ruleset; this module compares the two
4sets of findings and classifies each rule's change:
5
6* **regressed** — a control that was supported (or not yet observed) now fails
7* **fixed** — a control that failed now passes (or is no longer observed)
8* **drifted** — an ongoing failure whose failing evidence rows changed
9* **changed** — a non-failure status change (e.g. a table stopped being collected)
10* **unchanged** — same status, same evidence
11
12Evidence rows are compared as whole rows, so drift shows exactly which items
13appeared or disappeared (a new MFA-less user, a security group that was closed).
14"""
15
16from __future__ import annotations
17
18from dataclasses import dataclass, field
19from datetime import datetime, timezone
20from html import escape
21
22from .engine import FAIL, Finding
23from .loader import Package
24from .reporters.html import CSS as _CSS
25
26ABSENT = "absent" # rule present in only one of the two packages
27
28REGRESSED = "regressed"
29FIXED = "fixed"
30DRIFTED = "drifted"
31CHANGED = "changed"
32UNCHANGED = "unchanged"
33
34# Order categories appear in a report and how they roll up in the summary.
35CATEGORY_ORDER = [REGRESSED, FIXED, DRIFTED, CHANGED, UNCHANGED]
36_STATUS_LABEL = {FAIL: "fail", "pass": "pass", "not_applicable": "n/a", ABSENT: "absent"}
37_SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2}
38
39
40def _row_key(row: dict[str, str]) -> tuple:
41 return tuple(sorted(row.items()))
42
43
44@dataclass
45class RuleDelta:
46 """How one rule's finding changed between the two packages."""
47
48 rule: object # audit_report.rules.Rule
49 old_status: str
50 new_status: str
51 category: str
52 new_reason: str = ""
53 evidence_added: list[dict[str, str]] = field(default_factory=list)
54 evidence_removed: list[dict[str, str]] = field(default_factory=list)
55
56 @property
57 def controls(self) -> list[str]:
58 return self.rule.controls
59
60
61def _classify(old_status, new_status, added, removed) -> str:
62 if old_status == ABSENT:
63 return REGRESSED if new_status == FAIL else CHANGED
64 if new_status == ABSENT:
65 return CHANGED # rule dropped from the current ruleset
66 if new_status == FAIL and old_status != FAIL:
67 return REGRESSED
68 if old_status == FAIL and new_status != FAIL:
69 return FIXED
70 if old_status == FAIL and new_status == FAIL:
71 return DRIFTED if (added or removed) else UNCHANGED
72 return CHANGED if old_status != new_status else UNCHANGED
73
74
75def diff_findings(old: list[Finding], new: list[Finding]) -> list[RuleDelta]:
76 """Compare two finding lists (same ruleset) into a list of deltas."""
77 old_by_id = {f.rule.id: f for f in old}
78 new_by_id = {f.rule.id: f for f in new}
79
80 # New order first (ruleset order), then any rules only the baseline had.
81 ordered_ids = [f.rule.id for f in new]
82 ordered_ids += [f.rule.id for f in old if f.rule.id not in new_by_id]
83
84 deltas: list[RuleDelta] = []
85 for rid in ordered_ids:
86 of, nf = old_by_id.get(rid), new_by_id.get(rid)
87 rule = (nf or of).rule
88 old_status = of.status if of else ABSENT
89 new_status = nf.status if nf else ABSENT
90
91 old_ev = {_row_key(r): r for r in (of.evidence if of else [])}
92 new_ev = {_row_key(r): r for r in (nf.evidence if nf else [])}
93 added = [r for k, r in new_ev.items() if k not in old_ev]
94 removed = [r for k, r in old_ev.items() if k not in new_ev]
95
96 category = _classify(old_status, new_status, added, removed)
97 deltas.append(
98 RuleDelta(
99 rule=rule,
100 old_status=old_status,
101 new_status=new_status,
102 category=category,
103 new_reason=nf.reason if nf else "",
104 evidence_added=added,
105 evidence_removed=removed,
106 )
107 )
108 return deltas
109
110
111@dataclass
112class DiffReport:
113 """A computed comparison of two packages, ready to render."""
114
115 baseline: Package
116 current: Package
117 deltas: list[RuleDelta]
118 generated_at: str
119
120 def by_category(self, category: str) -> list[RuleDelta]:
121 rows = [d for d in self.deltas if d.category == category]
122 return sorted(rows, key=lambda d: _SEVERITY_ORDER.get(d.rule.severity, 1))
123
124 @property
125 def counts(self) -> dict[str, int]:
126 return {cat: len(self.by_category(cat)) for cat in CATEGORY_ORDER}
127
128
129def build_diff(
130 baseline: Package,
131 current: Package,
132 old_findings: list[Finding],
133 new_findings: list[Finding],
134) -> DiffReport:
135 """Assemble a :class:`DiffReport` with a UTC timestamp."""
136 stamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
137 return DiffReport(
138 baseline=baseline,
139 current=current,
140 deltas=diff_findings(old_findings, new_findings),
141 generated_at=stamp,
142 )
143
144
145def has_regression(diff: DiffReport, threshold: str) -> bool:
146 """True if any regressed rule meets the severity *threshold* ('none' = off)."""
147 if threshold == "none":
148 return False
149 floor = _SEVERITY_ORDER[threshold]
150 return any(
151 _SEVERITY_ORDER.get(d.rule.severity, 1) <= floor
152 for d in diff.by_category(REGRESSED)
153 )
154
155
156# --------------------------------------------------------------------------- #
157# Rendering
158# --------------------------------------------------------------------------- #
159
160_CATEGORY_HEADING = {
161 REGRESSED: "Regressions — a control is no longer supported",
162 FIXED: "Fixed — a previously failing control now passes",
163 DRIFTED: "Ongoing failures with changed evidence",
164 CHANGED: "Other status changes",
165 UNCHANGED: "Unchanged",
166}
167
168
169def _transition(delta: RuleDelta) -> str:
170 return f"{_STATUS_LABEL.get(delta.old_status, delta.old_status)} → {_STATUS_LABEL.get(delta.new_status, delta.new_status)}"
171
172
173def _md_table(rows: list[dict[str, str]]) -> list[str]:
174 if not rows:
175 return []
176 shown = rows[:10]
177 headers = list(shown[0].keys())
178 out = [
179 "| " + " | ".join(headers) + " |",
180 "| " + " | ".join("---" for _ in headers) + " |",
181 ]
182 out += ["| " + " | ".join(str(r.get(h, "")) for h in headers) + " |" for r in shown]
183 if len(rows) > len(shown):
184 out.append(f"\n_+{len(rows) - len(shown)} more row(s) omitted._")
185 return out
186
187
188def _render_md(diff: DiffReport) -> str:
189 counts = diff.counts
190 out: list[str] = []
191 out.append(f"# Evidence Drift — {diff.current.subject} ({diff.current.platform})")
192 out.append("")
193 out.append(f"- **Baseline:** `{diff.baseline.path.name}`")
194 out.append(f"- **Current:** `{diff.current.path.name}`")
195 out.append(f"- **Generated:** {diff.generated_at}")
196 out.append(
197 f"- **Drift:** {counts[REGRESSED]} regressed · {counts[FIXED]} fixed · "
198 f"{counts[DRIFTED]} drifted · {counts[CHANGED]} changed · "
199 f"{counts[UNCHANGED]} unchanged"
200 )
201 out.append("")
202 out.append(
203 "> A *regression* means a setting moved into a state that no longer "
204 "supports a control since the baseline. As always this is evidence, not "
205 "a verdict."
206 )
207 out.append("")
208
209 for category in CATEGORY_ORDER:
210 rows = diff.by_category(category)
211 if not rows or category == UNCHANGED:
212 continue
213 out.append(f"## {_CATEGORY_HEADING[category]}")
214 out.append("")
215 for delta in rows:
216 rule = delta.rule
217 out.append(f"### {rule.title}")
218 out.append("")
219 out.append(f"- **Rule:** `{rule.id}` · **Severity:** {rule.severity}")
220 out.append(f"- **Controls:** {', '.join(rule.controls) or '—'}")
221 out.append(f"- **Change:** {_transition(delta)}")
222 if delta.new_reason:
223 out.append(f"- **Now:** {delta.new_reason}")
224 if category == REGRESSED and rule.remediation:
225 out.append(f"- **Remediation:** {rule.remediation.strip()}")
226 out.append("")
227 if delta.evidence_added:
228 out.append("**Newly failing rows:**")
229 out.append("")
230 out.extend(_md_table(delta.evidence_added))
231 out.append("")
232 if delta.evidence_removed:
233 out.append("**No longer failing rows:**")
234 out.append("")
235 out.extend(_md_table(delta.evidence_removed))
236 out.append("")
237
238 unchanged = diff.counts[UNCHANGED]
239 if unchanged:
240 out.append(f"_{unchanged} rule(s) unchanged._")
241 return "\n".join(out).rstrip() + "\n"
242
243
244def _html_table(rows: list[dict[str, str]], caption: str, cls: str) -> str:
245 shown = rows[:10]
246 headers = list(shown[0].keys())
247 head = "".join(f"<th>{escape(h)}</th>" for h in headers)
248 body = "".join(
249 "<tr>" + "".join(f"<td>{escape(str(r.get(h, '')))}</td>" for h in headers) + "</tr>"
250 for r in shown
251 )
252 return (
253 f"<table class='{cls}'><caption>{escape(caption)}</caption>"
254 f"<thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"
255 )
256
257
258_DIFF_CSS = (
259 _CSS
260 + """
261.delta { border: 1px solid #e5e5e5; border-radius: 6px; padding: 1rem 1.1rem; margin: .8rem 0; }
262.delta.regressed { border-left: 4px solid #c1272d; }
263.delta.fixed { border-left: 4px solid #1a7f37; }
264.delta.drifted { border-left: 4px solid #d08700; }
265.delta.changed { border-left: 4px solid #bbb; }
266.delta h3 { margin: 0 0 .4rem; font-size: 1.05rem; }
267.transition { font-weight: 700; }
268table.added caption, table.removed caption { text-align: left; font-weight: 600; font-size: .85rem; padding: .2rem 0; }
269table.added caption { color: #c1272d; } table.removed caption { color: #1a7f37; }
270@media (prefers-color-scheme: dark) {
271 .delta { border-color: #2d2e33; }
272 table.added caption { color: #ff6b70; } table.removed caption { color: #4ac36a; }
273}
274"""
275)
276
277
278def _render_html(diff: DiffReport) -> str:
279 counts = diff.counts
280 parts: list[str] = []
281 parts.append(
282 f"<h1>Evidence Drift — {escape(diff.current.subject)} "
283 f"({escape(diff.current.platform)})</h1>"
284 )
285 parts.append(
286 f"<p class='meta'>Baseline <code>{escape(diff.baseline.path.name)}</code> → "
287 f"Current <code>{escape(diff.current.path.name)}</code> · "
288 f"Generated {escape(diff.generated_at)}</p>"
289 )
290 parts.append(
291 "<p class='summary-pills'>"
292 f"<span>{counts[REGRESSED]} regressed</span>"
293 f"<span>{counts[FIXED]} fixed</span>"
294 f"<span>{counts[DRIFTED]} drifted</span>"
295 f"<span>{counts[CHANGED]} changed</span>"
296 f"<span>{counts[UNCHANGED]} unchanged</span></p>"
297 )
298 parts.append(
299 "<p class='note'>A <strong>regression</strong> means a setting moved into "
300 "a state that no longer supports a control since the baseline. As always "
301 "this is evidence, not a verdict.</p>"
302 )
303
304 for category in CATEGORY_ORDER:
305 rows = diff.by_category(category)
306 if not rows or category == UNCHANGED:
307 continue
308 parts.append(f"<h2>{escape(_CATEGORY_HEADING[category])}</h2>")
309 for delta in rows:
310 rule = delta.rule
311 body = [
312 f"<h3>{escape(rule.title)}</h3>",
313 (
314 f"<dl><dt>Rule</dt><dd><code>{escape(rule.id)}</code> · "
315 f"{escape(rule.severity)}</dd>"
316 ),
317 f"<dt>Controls</dt><dd>{escape(', '.join(rule.controls) or '—')}</dd>",
318 f"<dt>Change</dt><dd class='transition'>{escape(_transition(delta))}</dd>",
319 ]
320 if delta.new_reason:
321 body.append(f"<dt>Now</dt><dd>{escape(delta.new_reason)}</dd>")
322 if category == REGRESSED and rule.remediation:
323 body.append(f"<dt>Remediation</dt><dd>{escape(rule.remediation.strip())}</dd>")
324 body.append("</dl>")
325 if delta.evidence_added:
326 body.append(_html_table(delta.evidence_added, "Newly failing rows", "added"))
327 if delta.evidence_removed:
328 body.append(
329 _html_table(delta.evidence_removed, "No longer failing rows", "removed")
330 )
331 parts.append(f"<div class='delta {category}'>{''.join(body)}</div>")
332
333 if counts[UNCHANGED]:
334 parts.append(f"<p class='meta'>{counts[UNCHANGED]} rule(s) unchanged.</p>")
335 parts.append("<footer>Generated by audit-report · Audit Labs · evidence, not a verdict.</footer>")
336
337 return (
338 "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
339 "<meta name='viewport' content='width=device-width, initial-scale=1'>"
340 f"<title>Evidence Drift — {escape(diff.current.subject)}</title>"
341 f"<style>{_DIFF_CSS}</style></head><body><main>{''.join(parts)}</main></body></html>\n"
342 )
343
344
345def _render_json(diff: DiffReport) -> str:
346 import json as _json
347
348 payload = {
349 "subject": diff.current.subject,
350 "platform": diff.current.platform,
351 "baseline_package": diff.baseline.path.name,
352 "current_package": diff.current.path.name,
353 "generated_at": diff.generated_at,
354 "summary": diff.counts,
355 "deltas": [
356 {
357 "id": d.rule.id,
358 "title": d.rule.title,
359 "severity": d.rule.severity,
360 "controls": d.rule.controls,
361 "category": d.category,
362 "old_status": d.old_status,
363 "new_status": d.new_status,
364 "evidence_added": d.evidence_added,
365 "evidence_removed": d.evidence_removed,
366 }
367 for d in diff.deltas
368 ],
369 }
370 return _json.dumps(payload, indent=2) + "\n"
371
372
373_RENDERERS = {"md": _render_md, "html": _render_html, "json": _render_json}
374
375
376def render(diff: DiffReport, fmt: str) -> str:
377 """Render a diff in the named format ('md', 'html', or 'json')."""
378 try:
379 return _RENDERERS[fmt](diff)
380 except KeyError:
381 raise ValueError(f"unknown format: {fmt!r}") from None