Integrate metrics from metrics.json into the report pdf.
This commit is contained in:
1 parent
9d7df77923
commit
9247608a9f
5 files changed
+149
-12
No files matched your search
@@ -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}"
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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}
|
||||
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]
|
||||
return {
|
||||
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 {
|
||||
"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),
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -156,6 +156,10 @@
|
||||
<h2>{{ block.title }}</h2>
|
||||
{{ distribution(block.rows, block.legend) }}
|
||||
{% endfor %}
|
||||
|
||||
{% for metric in summary.metrics or [] %}
|
||||
{{ card(metric.title, metric.rows) }}
|
||||
{% endfor %}
|
||||
{% endif %}
|
||||
|
||||
{% if critical %}
|
||||
|
||||
Reference in new issue
Block a user