Files
pdf-reporting/report_generator/context.py
T

687 lines
27 KiB
Python

"""Turns a scan's stored artifacts into the context the report template draws.
No parsing of its own. The SARIF and the vulnerability report are read by the summarisers in
:mod:`cipherverifier.core.services.sarif` and :mod:`cipherverifier.core.services.vdr` -- both of
which are deliberately free of Django and of any store -- and the pipeline metadata by the helpers
in :mod:`persistent_storage.utils.rep_utils`. What is left here is the mapping: which summarised
number goes in which row, and what to say where the artifacts do not say anything.
That last part is the rule this module is built around. A field the artifacts cannot answer is
written as :data:`UNKNOWN` rather than guessed, left blank or quietly dropped. A reader who sees
``<xxx>`` knows the pipeline did not report it; a reader who sees a plausible zero does not.
Moving this to archon is a change of what :mod:`sources` reads
from. The two summarisers travel with it -- they belong beside whatever holds the artifacts, not
inside the web application that happens to import them today.
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from .vendor.cipherverifier.core.services.sarif import summarize_sarif
from .vendor.cipherverifier.core.services.vdr import summarize_vdr
if TYPE_CHECKING:
from .vendor.cipherverifier.core.services.sarif import Finding, SarifReport
from .vendor.cipherverifier.core.services.vdr import VdrReport, Vulnerability
from .sources import ReportSources
# What stands in for a field the stored artifacts do not answer.
UNKNOWN = "<xxx>"
# How many of each list the report carries. A report is read, not audited; the full set is in the
# artifacts it was built from.
MOST_CRITICAL = 6
CARDS_PER_GROUP = 8
BARS = 10
# The severity a SARIF level is reported under. SARIF names four levels and the report bands five,
# so the mapping is written down rather than inferred from the spelling.
SEVERITY_OF_LEVEL = {
"error": "Critical",
"warning": "High",
"note": "Medium",
"none": "Low",
"info": "Low",
}
SEVERITY_ORDER = ("Critical", "High", "Medium", "Low")
def build_context(sources: ReportSources, documents: dict[str, Any]) -> dict[str, Any]:
"""Build the report context from a scan's parsed artifacts.
Args:
sources: The scan and its stored files.
documents: The parsed artifacts by role -- ``metadata``, ``findings``,
``vulnerabilities`` -- each absent where the scan stored none.
Returns:
The context :func:`layout.render_pdf`
draws. Sections whose artifact is missing are left out entirely, which
the template renders as a shorter report.
"""
metadata = documents.get("metadata") or {}
sast = summarize_sarif(documents["findings"]) if documents.get("findings") else None
sca = summarize_vdr(documents["vulnerabilities"]) if documents.get("vulnerabilities") else None
context: dict[str, Any] = {
"cover": _cover(sources, metadata, sast, sca),
"scan": _scan(sources, metadata),
"project": _project(metadata),
"summary": _summary(sast, sca, documents.get("metrics")),
"sections": [],
}
critical = _critical(sast, sca)
if critical["table"]["rows"]:
context["critical"] = critical
if sast is not None:
context["sections"].append(_sast_section(sast))
if sca is not None:
context["sections"].append(_sca_section(sca))
return context
# --- the page ------------------------------------------------------------------------------
def _fact(label: str, value: Any, *, gap: bool = False, kind: str | None = None) -> dict[str, Any]: # noqa: ANN401
"""One label/value row, with anything unstated written as :data:`UNKNOWN`."""
return {"label": label, "value": _stated(value), "gap": gap, "kind": kind}
def _row(label: str, value: Any, **marks: Any) -> dict[str, Any]: # noqa: ANN401
"""One row of a card."""
return {"label": label, "value": _stated(value), **marks}
def _stated(value: Any) -> str: # noqa: ANN401
"""What a value reads as, or :data:`UNKNOWN` where there is nothing to read.
Zero is a reading and survives; an empty string and ``None`` are not, and do not. The difference
matters on a page where both appear: "0 filtered" is a measurement, and a blank is a gap.
"""
if value is None or (isinstance(value, str) and not value.strip()):
return UNKNOWN
return str(value)
def _cover(
sources: ReportSources,
metadata: dict[str, Any],
sast: SarifReport | None,
sca: VdrReport | None,
) -> dict[str, Any]:
"""The first page."""
configuration = metadata.get("configuration") or {}
technologies = [name for name, present in (("SAST", sast is not None), ("SCA", sca is not None)) if present]
return {
"scan_id": str(sources.scan_id),
"subject": _stated(configuration.get("repository_name")),
"footer_date": datetime.now(UTC).strftime("%d %B %Y"),
"logo": "cc_logo.png",
"title": "SAST/SCA Report",
"identity": [
_fact("Scan Identifier", sources.scan_id, kind="mono"),
# Neither exists in the store today: a scan has artifacts, and a report of it is not
# itself recorded as one.
_fact("Report ID", None, kind="mono"),
_fact("Artifact ID", None, kind="mono"),
_fact("Technology", ", ".join(technologies) or None),
_fact("Project", configuration.get("repository_name"), gap=True),
_fact("Version", configuration.get("repository_version")),
_fact("Report Created", datetime.now(UTC).strftime("%A, %B %d, %Y")),
_fact("Manufacturer", None),
],
"severity_ledger": _severity_ledger(sast, sca),
}
def _severity_ledger(sast: SarifReport | None, sca: VdrReport | None) -> dict[str, Any] | None:
"""How many issues of each severity survived filtering, over both halves of the pipeline."""
counts = dict.fromkeys(SEVERITY_ORDER, 0)
for finding in _true_findings(sast):
counts[SEVERITY_OF_LEVEL.get(finding.level, "Low")] += 1
for vulnerability in sca.vulnerabilities if sca else []:
counts[_vdr_severity(vulnerability)] = counts.get(_vdr_severity(vulnerability), 0) + 1
if not any(counts.values()):
return None
rows: list[dict[str, Any]] = [
{"label": f"{name} severity issues", "count": counts.get(name, 0)} for name in SEVERITY_ORDER
]
rows.append({"label": "Total true issues", "count": sum(counts.values()), "total": True})
return {"title": "True issues by severity", "rows": rows}
def _scan(sources: ReportSources, metadata: dict[str, Any]) -> dict[str, Any]:
"""What the pipeline did, stage by stage."""
stages = [stage for stage in metadata.get("stages") or [] if isinstance(stage, dict)]
failed = [stage["label"] for stage in stages if stage.get("status") not in ("success", None)]
return {
"title": "CipherCell Scan Information",
"overview": [
_fact("Scan started at", _moment(metadata.get("pipeline_start_iso"))),
_fact("Scan duration", _duration(metadata.get("pipeline_duration_seconds"))),
_fact("Scan completed at", _moment(metadata.get("pipeline_end_iso"))),
_fact(
"Pipeline Steps", ", ".join(_stated(stage.get("label") or stage.get("key")) for stage in stages) or None
),
_fact("Errors Detected", ", ".join(failed) if failed else "None" if stages else None),
],
"stages": [_stage(stage, sources) for stage in stages],
}
def _stage(stage: dict[str, Any], sources: ReportSources) -> dict[str, Any]:
"""One stage of the pipeline.
The tools a stage ran and the profile it ran them under are not recorded in the metadata, so
they are left unstated. What is recorded is when it ran, how it ended, and -- indirectly -- what
it produced, which is read off the names of the files the scan stored.
"""
return {
"title": stage.get("label") or stage.get("key") or UNKNOWN,
"rows": [
_fact("Selected Tools", None),
_fact("Profile", None),
_fact("Duration", _duration(stage.get("duration_seconds"))),
_fact("Created Artifacts", ", ".join(sorted(sources.files)) or None, kind="mono"),
_fact("Errors Detected", "None" if stage.get("status") == "success" else stage.get("status")),
],
}
def _project(metadata: dict[str, Any]) -> dict[str, Any]:
"""What was scanned."""
configuration = metadata.get("configuration") or {}
return {
"title": "Project Information",
"overview": [
_fact("Project", configuration.get("repository_name")),
_fact("Version", configuration.get("repository_version")),
# Nothing in the pipeline counts lines, so nothing here can report them.
_fact("Analysed lines of code", None),
],
# Left as one unstated row rather than a table of zeroes: no stage reports a language
# breakdown, and a share invented from the file extensions of the findings would be a
# measurement of where the bugs are, not of what the project is written in.
"languages": {"title": "Languages", "rows": [{"label": UNKNOWN, "share": UNKNOWN}]},
}
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
summary = {
"title": "Summary",
"overview": [
_fact("Raw findings", raw or None),
# Only SAST findings carry a filtering verdict. A vulnerability suppressed in
# Dependency-Track never reaches the report at all, so the two cannot be added up.
_fact("Filtered", f"{filtered} (SAST)" if sast else None),
_fact("True issues", f"{len(true_sast)} SAST, {sca_total} SCA" if sast or sca else None),
# Each half stated only where its artifact was there. A scan that ran no SCA stage
# reports its SAST rules and says nothing about components, rather than nothing at all.
_fact("Distinct issue types (true)", ", ".join(_issue_types(sast, sca)) or None),
],
"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",
"note": "CVSS is the mean base score of the scored dependency findings. Impact values are normalized averages from 0 to 10; higher means greater potential damage.",
"rows": [
_fact("Average CVSS score (0-10)", _metric(scores.get("cvss"), "mean")),
_fact("Average CVSS severity", (scores.get("cvss") or {}).get("band"), kind="severity"),
_fact("Average confidentiality impact (0-10)", _metric(impact.get("scores"), "confidentiality")),
_fact("Average integrity impact (0-10)", _metric(impact.get("scores"), "integrity")),
_fact("Average availability impact (0-10)", _metric(impact.get("scores"), "availability")),
],
},
{
"title": "Exploitability",
"note": "These are normalized averages of the CVSS vector dimensions across scored findings. Higher values indicate fewer obstacles for an attacker, not a probability of exploitation.",
"rows": [
_fact(
label,
_metric(exploitability.get(axis), "score"),
)
for axis, label in (
("attack_vector", "Average attack vector exposure (0-10)"),
("attack_complexity", "Average attack complexity exposure (0-10)"),
("privileges_required", "Average privilege exposure (0-10)"),
("user_interaction", "Average user-interaction exposure (0-10)"),
)
],
},
{
"title": "Temporal",
"note": "Published dates and vulnerability age are calculated from the advisory timestamps. Time-to-exploit and volume trends require historical exploitation or scan data that this export does not contain.",
"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",
"note": "Surface counts describe the packages represented by the findings. CWE coverage counts findings with a CWE classification; it does not measure CWE quality or abstraction depth.",
"rows": [
_fact("Affected dependencies", surface.get("products")),
_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 _issue_types(sast: SarifReport | None, sca: VdrReport | None) -> list[str]:
"""How many distinct kinds of issue each half of the pipeline found."""
counted = []
if sast is not None:
counted.append(f"{sast.rule_count} SAST rules")
if sca is not None:
counted.append(f"{sca.component_count} affected components")
return counted
def _distributions(findings: list[Finding], sca: VdrReport | None) -> list[dict[str, Any]]:
"""Where the issues concentrate: by rule for SAST, by package for SCA."""
blocks = []
if findings:
blocks.append(
{
"title": f"True SAST issues by rule ({len(findings)})",
**_bars(_group(findings, lambda one: one.rule_id or UNKNOWN, _finding_severity)),
}
)
if sca and sca.vulnerabilities:
blocks.append(
{
"title": f"Most affected components, true SCA issues ({sca.component_count})",
**_bars(
_group(
[(one, name) for one in sca.vulnerabilities for name in one.component_names],
lambda pair: pair[1],
lambda pair: _vdr_severity(pair[0]),
),
mono=True,
),
}
)
return blocks
def _group[T](
items: list[T],
label_of: Any, # noqa: ANN401
severity_of: Any, # noqa: ANN401
) -> list[tuple[str, dict[str, int]]]:
"""Count items by label, and within each label by severity."""
grouped: dict[str, dict[str, int]] = {}
for item in items:
counts = grouped.setdefault(label_of(item), dict.fromkeys(SEVERITY_ORDER, 0))
counts[severity_of(item)] += 1
return sorted(grouped.items(), key=lambda pair: -sum(pair[1].values()))
def _bars(grouped: list[tuple[str, dict[str, int]]], *, mono: bool = False) -> dict[str, Any]:
"""Turn grouped counts into bars scaled against the busiest row.
Only the busiest :data:`BARS` are drawn, and the caveat says so where the rest were left off.
Without that the legend counts what is on the page while the heading counts the report, and the
two quietly disagree by however much the tail held.
"""
shown = grouped[:BARS]
largest = max((sum(counts.values()) for _, counts in shown), default=0) or 1
rows: list[dict[str, Any]] = []
totals = dict.fromkeys(SEVERITY_ORDER, 0)
for label, counts in shown:
rows.append(
{
"label": label,
"count": sum(counts.values()),
"mono": mono,
"parts": [
{"severity": name, "share": round(counts[name] / largest * 100, 1)}
for name in SEVERITY_ORDER
if counts[name]
],
}
)
for name, count in counts.items():
totals[name] += count
return {
"rows": rows,
"legend": {
"entries": [
{"severity": name, "label": f"{name} ({totals[name]})"}
for name in reversed(SEVERITY_ORDER)
if totals[name]
],
"caveat": _caveat(shown, grouped, largest),
},
}
def _caveat(shown: list[Any], grouped: list[Any], largest: int) -> str:
"""What the bars leave out, and what their length is measured against."""
scale = f"bar length is proportional to the issue count (100 % = {largest})."
if len(shown) == len(grouped):
return scale
return f"{len(shown)} of {len(grouped)} shown; {scale}"
def _critical(sast: SarifReport | None, sca: VdrReport | None) -> dict[str, Any]:
"""The handful worth reading first, whichever half of the pipeline found them."""
rows: list[dict[str, Any]] = []
for finding in sorted(_true_findings(sast), key=lambda one: SEVERITY_ORDER.index(_finding_severity(one))):
rows.append(
{
"cells": [
"SAST",
{"value": _finding_severity(finding), "kind": "severity"},
_stated(finding.rule_id),
{"value": _stated(_where(finding)), "kind": "mono"},
_stated(finding.message),
]
}
)
for vulnerability in sorted(
sca.vulnerabilities if sca else [],
key=lambda one: SEVERITY_ORDER.index(_vdr_severity(one)),
):
rows.append(
{
"cells": [
"SCA",
{"value": _vdr_severity(vulnerability), "kind": "severity"},
_stated(vulnerability.id),
{"value": _stated(", ".join(vulnerability.component_names)), "kind": "mono"},
_stated(vulnerability.description),
]
}
)
return {
"title": "Most Critical Vulnerabilities",
"table": {
"columns": ["Type", "Severity", "CVE/CWE", "Affected File / Component", "Description"],
"rows": rows[:MOST_CRITICAL],
},
}
def _sast_section(sast: SarifReport) -> dict[str, Any]:
"""The SAST findings, split into what was kept and what was set aside."""
true_findings = _true_findings(sast)
filtered = [one for one in sast.findings if one.false_positive]
groups: list[dict[str, Any]] = []
if true_findings:
groups.append(
{
"title": f"True Issues ({len(true_findings)})",
"cards": [_finding_card(one) for one in true_findings[:CARDS_PER_GROUP]],
}
)
if filtered:
groups.append(
{
"title": f"Filtered Issues ({len(filtered)})",
"note": (
"The following findings were reported by the scanners but assessed as not exploitable in "
"this artifact. Each entry carries the verdict that filtered it."
),
"cards": [_finding_card(one, filtered_out=True) for one in filtered[:CARDS_PER_GROUP]],
"table": _filtered_table(filtered),
}
)
return {"title": "SAST Vulnerabilities", "groups": groups}
def _finding_card(finding: Finding, *, filtered_out: bool = False) -> dict[str, Any]:
"""One SARIF finding."""
rows = [
_row("Severity", _finding_severity(finding), kind="severity"),
_row("Location", _where(finding), kind="mono"),
_row("Description", finding.message),
_row("Reported By", ", ".join(finding.tools) or None),
]
if filtered_out:
rows.append(
_row(
"Filtering Reason",
finding.polygraph_reasoning,
aside=True,
token=finding.polygraph_verdict,
)
)
else:
# Polygraph's verdict where it looked at this one, and nothing where it did not: silence is
# not a clean bill of health, and must not be written as one.
rows.append(_row("Evidence", finding.verdict_label or None))
return {"title": f"{_stated(finding.rule_id)} — {_stated(finding.rule_name)}", "rows": rows}
def _filtered_table(findings: list[Finding]) -> dict[str, Any]:
"""How many of each rule were filtered, and on what grounds."""
grouped: dict[str, list[Finding]] = {}
for finding in findings:
grouped.setdefault(finding.rule_id or UNKNOWN, []).append(finding)
rows: list[dict[str, Any]] = [
{
"cells": [
rule,
len(group),
{"value": _stated(group[0].polygraph_verdict), "kind": "mono"},
]
}
for rule, group in sorted(grouped.items(), key=lambda pair: -len(pair[1]))
]
rows.append({"cells": ["Total", len(findings), ""], "total": True})
return {
"title": f"Filtered SAST issues by rule ({len(findings)})",
"columns": ["Rule", "Filtered", "Dominant verdict"],
"rows": rows,
}
def _sca_section(sca: VdrReport) -> dict[str, Any]:
"""The dependency findings."""
return {
"title": "SBOM / SCA Vulnerabilities",
"new_page": True,
"groups": [
{
"title": f"True Issues by CVEs ({sca.total})",
"cards": [_vulnerability_card(one) for one in sca.vulnerabilities[:CARDS_PER_GROUP]],
}
],
}
def _vulnerability_card(vulnerability: Vulnerability) -> dict[str, Any]:
"""One dependency finding, as much of it as the report stated."""
metrics = vulnerability.metrics
impact = [
f"{name} ({value})"
for name, value in (
("Confidentiality", metrics.confidentiality),
("Integrity", metrics.integrity),
("Availability", metrics.availability),
)
if value
]
return {
"title": _stated(vulnerability.id),
"rows": [
_row("Affected component", ", ".join(vulnerability.component_names) or None, kind="mono"),
_row("Severity", _vdr_severity(vulnerability), kind="severity"),
_row("Description", vulnerability.description),
_row("CWE", ", ".join(f"{one}: {vulnerability.cwe_names[one]}" for one in vulnerability.cwes) or None),
_row("CVSS", f"{vulnerability.score} / 10" if vulnerability.score is not None else None),
_row("Impact", ", ".join(impact) or None),
_row("Exploitation likelihood (EPSS)", _percent(metrics.epss)),
_row(
"Fixed in",
# The report names the newest version of the package, which is not the same as the
# version that fixes this advisory -- so it is labelled as what it is.
next(
(one.latest_version for one in vulnerability.components if one.is_outdated),
None,
),
kind="mono",
),
_row("Known for", f"{metrics.published_days_ago} days" if metrics.published_days_ago else None),
_row("Recommendation", vulnerability.recommendation),
],
}
# --- readings ------------------------------------------------------------------------------
def _true_findings(sast: SarifReport | None) -> list[Finding]:
"""The findings that survived filtering."""
return [one for one in sast.findings if not one.false_positive] if sast else []
def _finding_severity(finding: Finding) -> str:
"""The severity band a SARIF level falls in."""
return SEVERITY_OF_LEVEL.get(finding.level, "Low")
def _vdr_severity(vulnerability: Vulnerability) -> str:
"""The severity band a vulnerability reports, capitalised the way the report bands them."""
named = (vulnerability.severity or "").capitalize()
return named if named in SEVERITY_ORDER else "Low"
def _where(finding: Finding) -> str | None:
"""Where a finding points, as ``file:line``."""
if finding.location is None:
return None
line = finding.location.start_line
return f"{finding.location.file}:{line}" if line else finding.location.file
def _percent(share: float | None) -> str | None:
"""A probability as a percentage, at the precision it is worth stating."""
if share is None:
return None
return f"{share * 100:.2f} %" if share < 0.01 else f"{share * 100:.1f} %"
def _moment(stated: Any) -> str | None: # noqa: ANN401
"""An ISO timestamp as the report writes it, or nothing where it cannot be read."""
if not isinstance(stated, str):
return None
try:
return datetime.fromisoformat(stated).strftime("%A, %B %d, %Y %H:%M:%S (%Z)").replace("()", "").strip()
except ValueError:
return None
def _duration(seconds: Any) -> str | None: # noqa: ANN401
"""A number of seconds as ``HH:MM:SS``."""
if not isinstance(seconds, (int, float)) or isinstance(seconds, bool):
return None
whole = int(seconds)
return f"{whole // 3600:02d}:{whole % 3600 // 60:02d}:{whole % 60:02d}"