| @@ -0,0 +1,381 @@ |
| 1 | """Diff mode — compare two evidence packages and report drift. |
| 2 | |
| 3 | Both packages are evaluated with the same ruleset; this module compares the two |
| 4 | sets 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 | |
| 12 | Evidence rows are compared as whole rows, so drift shows exactly which items |
| 13 | appeared or disappeared (a new MFA-less user, a security group that was closed). |
| 14 | """ |
| 15 | |
| 16 | from __future__ import annotations |
| 17 | |
| 18 | from dataclasses import dataclass, field |
| 19 | from datetime import datetime, timezone |
| 20 | from html import escape |
| 21 | |
| 22 | from .engine import FAIL, Finding |
| 23 | from .loader import Package |
| 24 | from .reporters.html import CSS as _CSS |
| 25 | |
| 26 | ABSENT = "absent" # rule present in only one of the two packages |
| 27 | |
| 28 | REGRESSED = "regressed" |
| 29 | FIXED = "fixed" |
| 30 | DRIFTED = "drifted" |
| 31 | CHANGED = "changed" |
| 32 | UNCHANGED = "unchanged" |
| 33 | |
| 34 | # Order categories appear in a report and how they roll up in the summary. |
| 35 | CATEGORY_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 | |
| 40 | def _row_key(row: dict[str, str]) -> tuple: |
| 41 | return tuple(sorted(row.items())) |
| 42 | |
| 43 | |
| 44 | @dataclass |
| 45 | class 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 | |
| 61 | def _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 | |
| 75 | def 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 |
| 112 | class 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 | |
| 129 | def 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 | |
| 145 | def 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 | |
| 169 | def _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 | |
| 173 | def _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 | |
| 188 | def _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 | |
| 244 | def _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; } |
| 268 | table.added caption, table.removed caption { text-align: left; font-weight: 600; font-size: .85rem; padding: .2rem 0; } |
| 269 | table.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 | |
| 278 | def _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 | |
| 345 | def _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 | |
| 376 | def 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 |