diff --git a/report_generator/commands/generate.py b/report_generator/commands/generate.py index a50c9a0..bde6359 100644 --- a/report_generator/commands/generate.py +++ b/report_generator/commands/generate.py @@ -126,7 +126,8 @@ def _report(sources: ReportSources, path: Path | None, *, write_metrics: bool = Raises: PipelineError: The report cannot be built or the destination not written. """ - name, content = render(sources) + metrics = build_metrics(sources) + name, content = render(sources, metrics) destination = _destination(path, name) try: @@ -134,7 +135,7 @@ def _report(sources: ReportSources, path: Path | None, *, write_metrics: bool = destination.write_bytes(content) if write_metrics: (destination.parent / "metrics.json").write_text( - json.dumps(build_metrics(sources), indent=2), encoding="utf-8" + json.dumps(metrics, indent=2), encoding="utf-8" ) except OSError as error: message = f"Cannot write the report to '{destination}': {error}" diff --git a/report_generator/context.py b/report_generator/context.py index 71c5a71..1d51807 100644 --- a/report_generator/context.py +++ b/report_generator/context.py @@ -72,7 +72,7 @@ def build_context(sources: ReportSources, documents: dict[str, Any]) -> dict[str "cover": _cover(sources, metadata, sast, sca), "scan": _scan(sources, metadata), "project": _project(metadata), - "summary": _summary(sast, sca), + "summary": _summary(sast, sca, documents.get("metrics")), "sections": [], } @@ -226,14 +226,14 @@ def _project(metadata: dict[str, Any]) -> dict[str, Any]: } -def _summary(sast: SarifReport | None, sca: VdrReport | None) -> dict[str, Any]: +def _summary(sast: SarifReport | None, sca: VdrReport | None, metrics: dict[str, Any] | None) -> dict[str, Any]: """The counts, and where they concentrate.""" true_sast = _true_findings(sast) sca_total = sca.total if sca else 0 raw = (sast.total if sast else 0) + sca_total filtered = sast.false_positive_count if sast else 0 - return { + summary = { "title": "Summary", "overview": [ _fact("Raw findings", raw or None), @@ -248,6 +248,90 @@ def _summary(sast: SarifReport | None, sca: VdrReport | None) -> dict[str, Any]: "distributions": _distributions(true_sast, sca), } + dependencies = metrics.get("dependencies") if isinstance(metrics, dict) else None + if isinstance(dependencies, dict): + summary["metrics"] = _metric_cards(dependencies) + + return summary + + +def _metric_cards(dependencies: dict[str, Any]) -> list[dict[str, Any]]: + """Turn the metrics document into compact summary cards.""" + scores = dependencies.get("scores") or {} + impact = dependencies.get("impact") or {} + exploitability = dependencies.get("exploitability") or {} + temporal = dependencies.get("temporal") or {} + published = temporal.get("published") or {} + remediation = dependencies.get("remediation") or {} + upgrades = remediation.get("newer_version_available") or {} + surface = dependencies.get("affected_surface") or {} + taxonomy = dependencies.get("taxonomy") or {} + + return [ + { + "title": "Severity & impact", + "rows": [ + _fact("CVSS", _metric(scores.get("cvss"), "mean")), + _fact("Severity band", (scores.get("cvss") or {}).get("band"), kind="severity"), + _fact("Confidentiality", _metric(impact.get("scores"), "confidentiality")), + _fact("Integrity", _metric(impact.get("scores"), "integrity")), + _fact("Availability", _metric(impact.get("scores"), "availability")), + ], + }, + { + "title": "Exploitability", + "rows": [_fact(_label(axis), _metric(exploitability.get(axis), "score")) for axis in ( + "attack_vector", "attack_complexity", "privileges_required", "user_interaction" + )], + }, + { + "title": "Temporal", + "rows": [ + _fact("Newest published", _published(published.get("newest"))), + _fact("Mean vulnerability age", _days(published.get("mean_days"))), + _fact("Median vulnerability age", _days(published.get("median_days"))), + _fact("Time to exploit", temporal.get("time_to_exploit_days")), + _fact("CVE volume trends", temporal.get("cve_volume_trends")), + ], + }, + { + "title": "Surface & taxonomy", + "rows": [ + _fact("Affected products", surface.get("products")), + _fact("Affected versions", surface.get("versions")), + _fact("Vendors", ", ".join(surface.get("vendors") or []) or None), + _fact("Potential remedies", _remedy(upgrades)), + _fact("CWE coverage", taxonomy.get("coverage")), + _fact("CWE abstraction level", taxonomy.get("abstraction_level")), + _fact("CWE relationships", taxonomy.get("relationships")), + ], + }, + ] + + +def _metric(value: Any, key: str) -> Any: # noqa: ANN401 + return value.get(key) if isinstance(value, dict) else None + + +def _published(value: Any) -> Any: # noqa: ANN401 + return value.get("cve") if isinstance(value, dict) else None + + +def _days(value: Any) -> str | None: # noqa: ANN401 + return f"{value} days" if isinstance(value, int | float) else None + + +def _remedy(value: dict[str, Any]) -> str | None: + findings = value.get("findings") + packages = value.get("packages") + if isinstance(findings, int) and isinstance(packages, int): + return f"{findings} findings across {packages} packages upgradeable" + return None + + +def _label(value: str) -> str: + return value.replace("_", " ").title() + def _issue_types(sast: SarifReport | None, sca: VdrReport | None) -> list[str]: """How many distinct kinds of issue each half of the pipeline found.""" diff --git a/report_generator/metrics.py b/report_generator/metrics.py index 1fe7f92..a6e1dd3 100644 --- a/report_generator/metrics.py +++ b/report_generator/metrics.py @@ -123,7 +123,9 @@ def _dependency_metrics(readings: list[dict[str, Any]]) -> dict[str, Any]: "impact": _impact(scored), "exploitability": _exploitability(scored, readings), "remediation": _remediation(readings), - "age": _age(readings), + "temporal": _temporal(readings), + "taxonomy": _taxonomy(readings), + "affected_surface": _affected_surface(readings), } @@ -255,7 +257,7 @@ def _remediation(readings: list[dict[str, Any]]) -> dict[str, Any]: } -def _age(readings: list[dict[str, Any]]) -> dict[str, Any]: +def _temporal(readings: list[dict[str, Any]]) -> dict[str, Any]: dated = [] for one in readings: stated = one["finding"].get("published") @@ -265,13 +267,58 @@ def _age(readings: list[dict[str, Any]]) -> dict[str, Any]: except ValueError: continue if not dated: - return {"newest": None, "oldest": None, "mean_days": None, "median_days": None, "coverage": 0} - dated.sort() - days = [(datetime.now(UTC) - date.astimezone(UTC)).days for date, _ in dated] + age = {"newest": None, "oldest": None, "mean_days": None, "median_days": None, "coverage": 0} + else: + dated.sort() + days = [(datetime.now(UTC) - date.astimezone(UTC)).days for date, _ in dated] + age = { + "newest": {"cve": dated[-1][1], "published": dated[-1][0].isoformat(), "days": days[-1]}, + "oldest": {"cve": dated[0][1], "published": dated[0][0].isoformat(), "days": days[0]}, + "mean_days": round(statistics.fmean(days)), + "median_days": round(statistics.median(days)), + "coverage": len(dated), + } + + return {"published": age, "time_to_exploit_days": None, "cve_volume_trends": None} + + +def _taxonomy(readings: list[dict[str, Any]]) -> dict[str, Any]: + cwes: Counter[str] = Counter() + for one in readings: + vulnerability = one["finding"] + values = vulnerability.get("cwes") or [] + if not values and vulnerability.get("cweId") is not None: + values = [vulnerability["cweId"]] + for value in values: + if isinstance(value, dict): + value = value.get("cweId") + if value is not None: + cwes[str(value) if str(value).startswith("CWE-") else f"CWE-{value}"] += 1 + return { - "newest": {"cve": dated[-1][1], "published": dated[-1][0].isoformat(), "days": days[-1]}, - "oldest": {"cve": dated[0][1], "published": dated[0][0].isoformat(), "days": days[0]}, - "mean_days": round(statistics.fmean(days)), - "median_days": round(statistics.median(days)), - "coverage": len(dated), + "cwe_counts": dict(cwes.most_common()), + "coverage": sum(cwes.values()), + "abstraction_level": None, + "relationships": None, + } + + +def _affected_surface(readings: list[dict[str, Any]]) -> dict[str, Any]: + components = [one["component"] for one in readings] + names = {str(component.get("name")) for component in components if component.get("name")} + versions = {str(component.get("version")) for component in components if component.get("version")} + vendors = set() + for component in components: + purl = component.get("purl") + if isinstance(purl, str) and purl.startswith("pkg:"): + coordinate = purl[4:].split("@", 1)[0] + parts = coordinate.split("/") + if len(parts) > 2 and parts[1]: + vendors.add(parts[1]) + + return { + "products": len(names), + "versions": len(versions), + "version_ranges": None, + "vendors": sorted(vendors), } \ No newline at end of file diff --git a/report_generator/rendering.py b/report_generator/rendering.py index eaa3d93..e087cb5 100644 --- a/report_generator/rendering.py +++ b/report_generator/rendering.py @@ -13,8 +13,8 @@ from __future__ import annotations import json from typing import TYPE_CHECKING, Any -from cc_common.api import PipelineError -from cc_common.api import context as logging_context +from .vendor.cc_common.api import PipelineError +from .vendor.cc_common.api import context as logging_context from .context import build_context from .layout import render_pdf @@ -40,7 +40,7 @@ REQUIRED = ("metadata",) NAME = "report_{scan}.pdf" -def render(sources: ReportSources) -> tuple[str, bytes]: +def render(sources: ReportSources, metrics: dict[str, Any] | None = None) -> tuple[str, bytes]: """Render the report of one scan. Args: @@ -56,6 +56,7 @@ def render(sources: ReportSources) -> tuple[str, bytes]: stage of their pipeline did not finish. """ documents = {name: _parsed(sources, name, candidates) for name, candidates in INPUTS.items()} + documents["metrics"] = metrics missing = [name for name in REQUIRED if documents.get(name) is None] if missing: diff --git a/report_generator/template/report.html.j2 b/report_generator/template/report.html.j2 index a636cf1..41451d7 100644 --- a/report_generator/template/report.html.j2 +++ b/report_generator/template/report.html.j2 @@ -156,6 +156,10 @@