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
main: 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_delta_md(delta, category: str) -> list[str]:
189 rule = delta.rule
190 out = [
191 f"### {rule.title}",
192 "",
193 f"- **Rule:** `{rule.id}` · **Severity:** {rule.severity}",
194 f"- **Controls:** {', '.join(rule.controls) or '—'}",
195 f"- **Change:** {_transition(delta)}",
196 ]
197 if delta.new_reason:
198 out.append(f"- **Now:** {delta.new_reason}")
199 if category == REGRESSED and rule.remediation:
200 out.append(f"- **Remediation:** {rule.remediation.strip()}")
201 out.append("")
202 if delta.evidence_added:
203 out.append("**Newly failing rows:**")
204 out.append("")
205 out.extend(_md_table(delta.evidence_added))
206 out.append("")
207 if delta.evidence_removed:
208 out.append("**No longer failing rows:**")
209 out.append("")
210 out.extend(_md_table(delta.evidence_removed))
211 out.append("")
212 return out
213
214
215def _render_md(diff: DiffReport) -> str:
216 counts = diff.counts
217 out: list[str] = []
218 out.append(f"# Evidence Drift — {diff.current.subject} ({diff.current.platform})")
219 out.append("")
220 out.append(f"- **Baseline:** `{diff.baseline.path.name}`")
221 out.append(f"- **Current:** `{diff.current.path.name}`")
222 out.append(f"- **Generated:** {diff.generated_at}")
223 out.append(
224 f"- **Drift:** {counts[REGRESSED]} regressed · {counts[FIXED]} fixed · "
225 f"{counts[DRIFTED]} drifted · {counts[CHANGED]} changed · "
226 f"{counts[UNCHANGED]} unchanged"
227 )
228 out.append("")
229 out.append(
230 "> A *regression* means a setting moved into a state that no longer "
231 "supports a control since the baseline. As always this is evidence, not "
232 "a verdict."
233 )
234 out.append("")
235
236 for category in CATEGORY_ORDER:
237 rows = diff.by_category(category)
238 if not rows or category == UNCHANGED:
239 continue
240 out.append(f"## {_CATEGORY_HEADING[category]}")
241 out.append("")
242 for delta in rows:
243 out.extend(_render_delta_md(delta, category))
244
245 unchanged = diff.counts[UNCHANGED]
246 if unchanged:
247 out.append(f"_{unchanged} rule(s) unchanged._")
248 return "\n".join(out).rstrip() + "\n"
249
250
251def _html_table(rows: list[dict[str, str]], caption: str, cls: str) -> str:
252 shown = rows[:10]
253 headers = list(shown[0].keys())
254 head = "".join(f"<th>{escape(h)}</th>" for h in headers)
255 body = "".join(
256 "<tr>" + "".join(f"<td>{escape(str(r.get(h, '')))}</td>" for h in headers) + "</tr>"
257 for r in shown
258 )
259 return (
260 f"<table class='{cls}'><caption>{escape(caption)}</caption>"
261 f"<thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"
262 )
263
264
265_DIFF_CSS = (
266 _CSS
267 + """
268.delta { border: 1px solid #e5e5e5; border-radius: 6px; padding: 1rem 1.1rem; margin: .8rem 0; }
269.delta.regressed { border-left: 4px solid #c1272d; }
270.delta.fixed { border-left: 4px solid #1a7f37; }
271.delta.drifted { border-left: 4px solid #d08700; }
272.delta.changed { border-left: 4px solid #bbb; }
273.delta h3 { margin: 0 0 .4rem; font-size: 1.05rem; }
274.transition { font-weight: 700; }
275table.added caption, table.removed caption { text-align: left; font-weight: 600; font-size: .85rem; padding: .2rem 0; }
276table.added caption { color: #c1272d; } table.removed caption { color: #1a7f37; }
277@media (prefers-color-scheme: dark) {
278 .delta { border-color: #2d2e33; }
279 table.added caption { color: #ff6b70; } table.removed caption { color: #4ac36a; }
280}
281"""
282)
283
284
285def _render_delta_html(delta, category: str) -> str:
286 rule = delta.rule
287 body = [
288 f"<h3>{escape(rule.title)}</h3>",
289 (
290 f"<dl><dt>Rule</dt><dd><code>{escape(rule.id)}</code> · "
291 f"{escape(rule.severity)}</dd>"
292 ),
293 f"<dt>Controls</dt><dd>{escape(', '.join(rule.controls) or '—')}</dd>",
294 f"<dt>Change</dt><dd class='transition'>{escape(_transition(delta))}</dd>",
295 ]
296 if delta.new_reason:
297 body.append(f"<dt>Now</dt><dd>{escape(delta.new_reason)}</dd>")
298 if category == REGRESSED and rule.remediation:
299 body.append(f"<dt>Remediation</dt><dd>{escape(rule.remediation.strip())}</dd>")
300 body.append("</dl>")
301 if delta.evidence_added:
302 body.append(_html_table(delta.evidence_added, "Newly failing rows", "added"))
303 if delta.evidence_removed:
304 body.append(
305 _html_table(delta.evidence_removed, "No longer failing rows", "removed")
306 )
307 return f"<div class='delta {category}'>{''.join(body)}</div>"
308
309
310def _render_html(diff: DiffReport) -> str:
311 counts = diff.counts
312 parts: list[str] = []
313 parts.append(
314 f"<h1>Evidence Drift — {escape(diff.current.subject)} "
315 f"({escape(diff.current.platform)})</h1>"
316 )
317 parts.append(
318 f"<p class='meta'>Baseline <code>{escape(diff.baseline.path.name)}</code> → "
319 f"Current <code>{escape(diff.current.path.name)}</code> · "
320 f"Generated {escape(diff.generated_at)}</p>"
321 )
322 parts.append(
323 "<p class='summary-pills'>"
324 f"<span>{counts[REGRESSED]} regressed</span>"
325 f"<span>{counts[FIXED]} fixed</span>"
326 f"<span>{counts[DRIFTED]} drifted</span>"
327 f"<span>{counts[CHANGED]} changed</span>"
328 f"<span>{counts[UNCHANGED]} unchanged</span></p>"
329 )
330 parts.append(
331 "<p class='note'>A <strong>regression</strong> means a setting moved into "
332 "a state that no longer supports a control since the baseline. As always "
333 "this is evidence, not a verdict.</p>"
334 )
335
336 for category in CATEGORY_ORDER:
337 rows = diff.by_category(category)
338 if not rows or category == UNCHANGED:
339 continue
340 parts.append(f"<h2>{escape(_CATEGORY_HEADING[category])}</h2>")
341 for delta in rows:
342 parts.append(_render_delta_html(delta, category))
343
344 if counts[UNCHANGED]:
345 parts.append(f"<p class='meta'>{counts[UNCHANGED]} rule(s) unchanged.</p>")
346 parts.append("<footer>Generated by audit-report · Audit Labs · evidence, not a verdict.</footer>")
347
348 return (
349 "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
350 "<meta name='viewport' content='width=device-width, initial-scale=1'>"
351 f"<title>Evidence Drift — {escape(diff.current.subject)}</title>"
352 f"<style>{_DIFF_CSS}</style></head><body><main>{''.join(parts)}</main></body></html>\n"
353 )
354
355
356def _render_json(diff: DiffReport) -> str:
357 import json as _json
358
359 payload = {
360 "subject": diff.current.subject,
361 "platform": diff.current.platform,
362 "baseline_package": diff.baseline.path.name,
363 "current_package": diff.current.path.name,
364 "generated_at": diff.generated_at,
365 "summary": diff.counts,
366 "deltas": [
367 {
368 "id": d.rule.id,
369 "title": d.rule.title,
370 "severity": d.rule.severity,
371 "controls": d.rule.controls,
372 "category": d.category,
373 "old_status": d.old_status,
374 "new_status": d.new_status,
375 "evidence_added": d.evidence_added,
376 "evidence_removed": d.evidence_removed,
377 }
378 for d in diff.deltas
379 ],
380 }
381 return _json.dumps(payload, indent=2) + "\n"
382
383
384_RENDERERS = {"md": _render_md, "html": _render_html, "json": _render_json}
385
386
387def render(diff: DiffReport, fmt: str) -> str:
388 """Render a diff in the named format ('md', 'html', or 'json')."""
389 try:
390 return _RENDERERS[fmt](diff)
391 except KeyError:
392 raise ValueError(f"unknown format: {fmt!r}") from None