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/trend.py · raw
1"""Trend mode — track each rule across a series of dated packages.
2
3Given a folder of audit-tools packages for the same platform and subject
4(``aws_audit_prod_2026-01-01/``, ``…_2026-02-01/``, …), this evaluates every
5package with the same ruleset and lays the results out as a timeline: one row
6per rule, one column per package date, so you can see a control drift in and out
7of compliance over time.
8"""
9
10from __future__ import annotations
11
12from dataclasses import dataclass
13from datetime import datetime, timezone
14from html import escape
15from pathlib import Path
16
17from .engine import FAIL, NOT_APPLICABLE, PASS, Finding
18from .loader import Package, detect_platform, package_date
19from .reporters.html import CSS as _CSS
20
21_STATUS_SYMBOL = {PASS: "✓", FAIL: "✗", NOT_APPLICABLE: "·"}
22_STATUS_CLASS = {PASS: "pass", FAIL: "fail", NOT_APPLICABLE: "na"}
23_SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2}
24
25
26def discover(parent: str | Path, subject: str | None = None) -> tuple[str, str, list[Path]]:
27 """Find a single series of packages under *parent*.
28
29 Returns ``(platform, subject, [paths sorted by date])``. Raises ``ValueError``
30 if no packages are found, if fewer than two share a platform/subject, or if
31 several distinct series are present and *subject* does not narrow it to one.
32 """
33 directory = Path(parent)
34 if not directory.is_dir():
35 raise ValueError(f"not a directory: {directory}")
36
37 groups: dict[tuple[str, str], list[Path]] = {}
38 for child in sorted(directory.iterdir()):
39 if not child.is_dir() or not any(child.glob("*.csv")):
40 continue
41 platform, subj = detect_platform(child.name)
42 if platform == "unknown":
43 continue
44 if subject and subj != subject:
45 continue
46 groups.setdefault((platform, subj), []).append(child)
47
48 if not groups:
49 raise ValueError(
50 f"no audit-tools packages found under {directory}"
51 + (f" for subject '{subject}'" if subject else "")
52 )
53 if len(groups) > 1:
54 listed = ", ".join(f"{p}/{s}" for p, s in sorted(groups))
55 raise ValueError(
56 f"multiple series found ({listed}); narrow with --subject and a "
57 "directory that holds one platform"
58 )
59
60 (platform, subj), paths = next(iter(groups.items()))
61 if len(paths) < 2:
62 raise ValueError("a trend needs at least two packages in the series")
63
64 paths.sort(key=lambda p: (package_date(p.name), p.name))
65 return platform, subj, paths
66
67
68@dataclass
69class TrendRow:
70 """One rule's status across the timeline."""
71
72 rule: object # audit_report.rules.Rule
73 statuses: list[str]
74
75 @property
76 def transitions(self) -> int:
77 """How many times the status changed along the timeline."""
78 return sum(1 for a, b in zip(self.statuses, self.statuses[1:]) if a != b)
79
80
81@dataclass
82class TrendReport:
83 """Rules-over-time view of a package series."""
84
85 platform: str
86 subject: str
87 dates: list[str] # column labels (package date or dir name)
88 rows: list[TrendRow]
89 generated_at: str
90
91 def fails_per_date(self) -> list[int]:
92 return [
93 sum(1 for row in self.rows if row.statuses[i] == FAIL)
94 for i in range(len(self.dates))
95 ]
96
97
98def build_trend(
99 packages: list[Package], findings_per_package: list[list[Finding]]
100) -> TrendReport:
101 """Assemble a :class:`TrendReport` from aligned packages and findings."""
102 dates = [pkg.date or pkg.path.name for pkg in packages]
103
104 # Preserve rule order from the first package; align by rule id across dates.
105 order = [f.rule for f in findings_per_package[0]]
106 by_date = [{f.rule.id: f for f in findings} for findings in findings_per_package]
107
108 rows = [
109 TrendRow(
110 rule=rule,
111 statuses=[
112 col.get(rule.id).status if col.get(rule.id) else NOT_APPLICABLE
113 for col in by_date
114 ],
115 )
116 for rule in order
117 ]
118 stamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
119 return TrendReport(
120 platform=packages[0].platform,
121 subject=packages[0].subject,
122 dates=dates,
123 rows=rows,
124 generated_at=stamp,
125 )
126
127
128def latest_findings(findings_per_package: list[list[Finding]]) -> list[Finding]:
129 """The findings of the most recent package (for --fail-on gating)."""
130 return findings_per_package[-1]
131
132
133# --------------------------------------------------------------------------- #
134# Rendering
135# --------------------------------------------------------------------------- #
136
137
138def _render_md(trend: TrendReport) -> str:
139 out: list[str] = []
140 out.append(f"# Evidence Trend — {trend.subject} ({trend.platform})")
141 out.append("")
142 out.append(f"- **Packages:** {len(trend.dates)}")
143 out.append(f"- **Timeline:** {trend.dates[0]} → {trend.dates[-1]}")
144 out.append(f"- **Generated:** {trend.generated_at}")
145 out.append("")
146 out.append("Legend: ✓ pass · ✗ fail · · not applicable")
147 out.append("")
148
149 header = "| Rule | Sev | " + " | ".join(trend.dates) + " |"
150 sep = "| --- | --- | " + " | ".join("---" for _ in trend.dates) + " |"
151 out.append(header)
152 out.append(sep)
153 for row in sorted(trend.rows, key=lambda r: _SEVERITY_ORDER.get(r.rule.severity, 1)):
154 cells = " | ".join(_STATUS_SYMBOL.get(s, "?") for s in row.statuses)
155 out.append(f"| {row.rule.title} | {row.rule.severity} | {cells} |")
156
157 fails = trend.fails_per_date()
158 out.append("| **Failing total** | | " + " | ".join(str(n) for n in fails) + " |")
159 return "\n".join(out).rstrip() + "\n"
160
161
162_TREND_CSS = (
163 _CSS
164 + """
165.trend { border-collapse: collapse; }
166.trend th.date { font-size: .78rem; white-space: nowrap; }
167.trend td.cell { text-align: center; font-weight: 700; width: 2.4rem; }
168.trend td.cell.pass { background: #e5f6ea; color: #1a7f37; }
169.trend td.cell.fail { background: #fdeaea; color: #c1272d; }
170.trend td.cell.na { background: #f4f4f6; color: #999; }
171.trend tr.totals td { font-weight: 700; background: #fafafa; }
172.rulecol { max-width: 22rem; }
173.legend { font-size: .85rem; color: #555; margin: .25rem 0 1rem; }
174@media (prefers-color-scheme: dark) {
175 .trend td.cell.pass { background: #12321d; color: #4ac36a; }
176 .trend td.cell.fail { background: #3a1416; color: #ff6b70; }
177 .trend td.cell.na { background: #202126; color: #888; }
178 .trend tr.totals td { background: #1c1d21; }
179 .legend { color: #aaa; }
180}
181"""
182)
183
184
185def _render_html(trend: TrendReport) -> str:
186 date_heads = "".join(f"<th class='date'>{escape(d)}</th>" for d in trend.dates)
187 body_rows: list[str] = []
188 for row in sorted(trend.rows, key=lambda r: _SEVERITY_ORDER.get(r.rule.severity, 1)):
189 cells = "".join(
190 f"<td class='cell {_STATUS_CLASS.get(s, 'na')}' title='{escape(s)}'>"
191 f"{_STATUS_SYMBOL.get(s, '?')}</td>"
192 for s in row.statuses
193 )
194 body_rows.append(
195 f"<tr><td class='rulecol'>{escape(row.rule.title)}"
196 f"<br><code>{escape(row.rule.id)}</code></td>"
197 f"<td>{escape(row.rule.severity)}</td>{cells}</tr>"
198 )
199 totals = "".join(f"<td class='cell'>{n}</td>" for n in trend.fails_per_date())
200
201 table = (
202 "<table class='trend'><thead><tr><th class='rulecol'>Rule</th><th>Sev</th>"
203 f"{date_heads}</tr></thead><tbody>{''.join(body_rows)}"
204 f"<tr class='totals'><td>Failing total</td><td></td>{totals}</tr>"
205 "</tbody></table>"
206 )
207 return (
208 "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
209 "<meta name='viewport' content='width=device-width, initial-scale=1'>"
210 f"<title>Evidence Trend — {escape(trend.subject)}</title>"
211 f"<style>{_TREND_CSS}</style></head><body><main>"
212 f"<h1>Evidence Trend — {escape(trend.subject)} ({escape(trend.platform)})</h1>"
213 f"<p class='meta'>{len(trend.dates)} packages · {escape(trend.dates[0])} → "
214 f"{escape(trend.dates[-1])} · Generated {escape(trend.generated_at)}</p>"
215 "<p class='legend'>✓ pass · ✗ fail · · not applicable — cell colour tracks "
216 "each control over time.</p>"
217 f"{table}"
218 "<footer>Generated by audit-report · Audit Labs · evidence, not a verdict.</footer>"
219 "</main></body></html>\n"
220 )
221
222
223def _render_json(trend: TrendReport) -> str:
224 import json as _json
225
226 payload = {
227 "subject": trend.subject,
228 "platform": trend.platform,
229 "dates": trend.dates,
230 "generated_at": trend.generated_at,
231 "fails_per_date": trend.fails_per_date(),
232 "rules": [
233 {
234 "id": row.rule.id,
235 "title": row.rule.title,
236 "severity": row.rule.severity,
237 "controls": row.rule.controls,
238 "statuses": row.statuses,
239 }
240 for row in trend.rows
241 ],
242 }
243 return _json.dumps(payload, indent=2) + "\n"
244
245
246_RENDERERS = {"md": _render_md, "html": _render_html, "json": _render_json}
247
248
249def render(trend: TrendReport, fmt: str) -> str:
250 """Render a trend in the named format ('md', 'html', or 'json')."""
251 try:
252 return _RENDERERS[fmt](trend)
253 except KeyError:
254 raise ValueError(f"unknown format: {fmt!r}") from None