785 lines
30 KiB
Python
785 lines
30 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
|
|
|
|
import re
|
|
from collections import Counter
|
|
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
|
|
|
|
from .layout import TITLE
|
|
|
|
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 entries the summary lists carry. The sections themselves list every finding.
|
|
MOST_CRITICAL = 6
|
|
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")
|
|
|
|
# A full git commit SHA.
|
|
_COMMIT = re.compile(r"[0-9a-f]{40}")
|
|
|
|
|
|
def build_context(
|
|
sources: ReportSources,
|
|
documents: dict[str, Any],
|
|
*,
|
|
project: str | None = None,
|
|
manufacturer: str | None = None,
|
|
) -> 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.
|
|
project: The project name the caller supplied. Wins over the one in the metadata.
|
|
manufacturer: The manufacturer the caller supplied. The artifacts never name one.
|
|
|
|
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 {}
|
|
project = project or (metadata.get("configuration") or {}).get("repository_name")
|
|
tools = _tools(documents.get("findings"))
|
|
sca = summarize_vdr(documents["vulnerabilities"]) if documents.get("vulnerabilities") else None
|
|
|
|
context: dict[str, Any] = {
|
|
"cover": _cover(sources, metadata, tools, sca, project=project, manufacturer=manufacturer),
|
|
"summary": _summary(tools, sca, documents.get("metrics")),
|
|
"sections": [_tool_section(name, report) for name, report in tools],
|
|
}
|
|
|
|
# Without the pipeline's metadata these two pages would be nothing but gaps.
|
|
if metadata:
|
|
context["scan"] = _scan(sources, metadata)
|
|
context["project"] = _project(metadata, project)
|
|
|
|
critical = _critical(tools, sca)
|
|
|
|
if critical["table"]["rows"]:
|
|
context["critical"] = critical
|
|
|
|
if sca is not None:
|
|
context["sections"].append(_sca_section(sca))
|
|
|
|
set_aside = _set_aside_section(tools)
|
|
|
|
if set_aside is not None:
|
|
context["sections"].append(set_aside)
|
|
|
|
context["contents"] = _contents(context)
|
|
|
|
return context
|
|
|
|
|
|
def _tools(sarif: Any) -> list[tuple[str, SarifReport]]: # noqa: ANN401
|
|
"""The SARIF document split by the tool that produced each run, in the order the runs appear."""
|
|
if not isinstance(sarif, dict):
|
|
return []
|
|
|
|
runs = [run for run in sarif.get("runs") or [] if isinstance(run, dict)]
|
|
|
|
return [(_tool_name(run), summarize_sarif({"runs": [run]})) for run in runs]
|
|
|
|
|
|
def _tool_name(run: dict[str, Any]) -> str:
|
|
"""What the tool that produced a run calls itself."""
|
|
driver = (run.get("tool") or {}).get("driver") or {}
|
|
|
|
return str(driver.get("name") or driver.get("fullName") or UNKNOWN)
|
|
|
|
|
|
def _contents(context: dict[str, Any]) -> list[dict[str, str]]:
|
|
"""The table of contents: every top-level part of the report that is actually in it."""
|
|
parts = [context[key] for key in ("scan", "project", "summary", "critical") if key in context]
|
|
|
|
return [{"id": part["id"], "title": part["title"]} for part in [*parts, *context["sections"]]]
|
|
|
|
|
|
# --- 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],
|
|
tools: list[tuple[str, SarifReport]],
|
|
sca: VdrReport | None,
|
|
*,
|
|
project: str | None,
|
|
manufacturer: str | None,
|
|
) -> dict[str, Any]:
|
|
"""The first page."""
|
|
technologies = [name for name, _ in tools] + (["SCA"] if sca is not None else [])
|
|
|
|
return {
|
|
"scan_id": str(sources.scan_id),
|
|
"subject": _stated(project),
|
|
"footer_date": datetime.now(UTC).strftime("%d %B %Y"),
|
|
"logo": "cc_logo.png",
|
|
"title": TITLE,
|
|
"identity": [
|
|
_fact("Scan Identifier", sources.scan_id, kind="mono"),
|
|
_fact("Technology", ", ".join(technologies) or None),
|
|
_fact("Project", project, gap=True),
|
|
_fact("Version", _version(metadata), kind="mono"),
|
|
_fact("Report Created", datetime.now(UTC).strftime("%A, %B %d, %Y")),
|
|
_fact("Manufacturer", manufacturer),
|
|
],
|
|
"severity_ledger": _severity_ledger(tools, sca),
|
|
}
|
|
|
|
|
|
def _severity_ledger(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> dict[str, Any] | None:
|
|
"""How many issues of each severity survived filtering, over every tool and the dependencies."""
|
|
counts = dict.fromkeys(SEVERITY_ORDER, 0)
|
|
|
|
for _, report in tools:
|
|
for finding in _kept(report):
|
|
counts[_finding_severity(finding)] += 1
|
|
|
|
for vulnerability in sca.vulnerabilities if sca else []:
|
|
counts[_vdr_severity(vulnerability)] += 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 {
|
|
"id": "scan",
|
|
"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 _version(metadata: dict[str, Any]) -> str | None:
|
|
"""What version of the project was scanned: the stated one, else the commit the SBOM was uploaded for.
|
|
|
|
The pipeline leaves ``repository_version`` empty but tags the Dependency-Track upload with the
|
|
commit it scanned, which is the most precise version there is.
|
|
"""
|
|
configuration = metadata.get("configuration") or {}
|
|
stated = configuration.get("repository_version")
|
|
|
|
if stated:
|
|
return str(stated)
|
|
|
|
commit = (configuration.get("deptrace_upload") or {}).get("project_version")
|
|
|
|
return commit if isinstance(commit, str) and _COMMIT.fullmatch(commit) else None
|
|
|
|
|
|
def _project(metadata: dict[str, Any], project: str | None) -> dict[str, Any]:
|
|
"""What was scanned."""
|
|
return {
|
|
"id": "project",
|
|
"title": "Project Information",
|
|
"overview": [
|
|
_fact("Project", project),
|
|
_fact("Version", _version(metadata), kind="mono"),
|
|
# 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(
|
|
tools: list[tuple[str, SarifReport]], sca: VdrReport | None, metrics: dict[str, Any] | None
|
|
) -> dict[str, Any]:
|
|
"""The counts, and where they concentrate."""
|
|
sca_total = sca.total if sca else 0
|
|
raw = sum(report.total for _, report in tools) + sca_total
|
|
true_issues = [f"{len(_kept(report))} {name}" for name, report in tools] + ([f"{sca_total} SCA"] if sca else [])
|
|
|
|
summary = {
|
|
"id": "summary",
|
|
"title": "Summary",
|
|
"overview": [
|
|
_fact("Raw findings", raw or None),
|
|
# Only SARIF findings can be set aside. A vulnerability suppressed in Dependency-Track
|
|
# never reaches the report at all, so the two cannot be added up.
|
|
_fact(
|
|
"Suppressed / False Positives",
|
|
", ".join(f"{len(_set_aside(report))} {name}" for name, report in tools) or None,
|
|
),
|
|
_fact("True issues", ", ".join(true_issues) or None),
|
|
_fact("Distinct issue types (true)", ", ".join(_issue_types(tools, sca)) or None),
|
|
],
|
|
"distributions": _distributions(tools, 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(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> list[str]:
|
|
"""How many distinct kinds of true issue each tool, and the dependency check, found."""
|
|
counted = [f"{len({one.rule_id for one in _kept(report)})} {name} rules" for name, report in tools]
|
|
|
|
if sca is not None:
|
|
counted.append(f"{sca.component_count} affected components")
|
|
|
|
return counted
|
|
|
|
|
|
def _distributions(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> list[dict[str, Any]]:
|
|
"""Where the issues concentrate: by rule for each SARIF tool, by package for SCA."""
|
|
blocks = []
|
|
|
|
for name, report in tools:
|
|
findings = _kept(report)
|
|
|
|
if findings:
|
|
blocks.append(
|
|
{
|
|
"title": f"True {name} 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(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> dict[str, Any]:
|
|
"""The handful worth reading first, whichever tool found them."""
|
|
rows: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
for name, report in tools:
|
|
for finding in _kept(report):
|
|
severity = _finding_severity(finding)
|
|
rows.append(
|
|
(
|
|
severity,
|
|
{
|
|
"cells": [
|
|
name,
|
|
{"value": severity, "kind": "severity"},
|
|
_stated(finding.rule_id),
|
|
{"value": _stated(_where(finding)), "kind": "mono"},
|
|
_stated(_text(finding.message)),
|
|
]
|
|
},
|
|
)
|
|
)
|
|
|
|
for vulnerability in sca.vulnerabilities if sca else []:
|
|
severity = _vdr_severity(vulnerability)
|
|
rows.append(
|
|
(
|
|
severity,
|
|
{
|
|
"cells": [
|
|
"SCA",
|
|
{"value": severity, "kind": "severity"},
|
|
_stated(vulnerability.id),
|
|
{"value": _stated(", ".join(vulnerability.component_names)), "kind": "mono"},
|
|
_stated(vulnerability.description),
|
|
]
|
|
},
|
|
)
|
|
)
|
|
|
|
# Sorted is stable, so within a severity each tool keeps its own order -- the rank, where it has one.
|
|
rows.sort(key=lambda pair: SEVERITY_ORDER.index(pair[0]))
|
|
|
|
return {
|
|
"id": "critical",
|
|
"title": "Most Critical Vulnerabilities",
|
|
"table": {
|
|
"columns": ["Source", "Severity", "CVE/CWE", "Affected File / Component", "Description"],
|
|
"rows": [row for _, row in rows[:MOST_CRITICAL]],
|
|
},
|
|
}
|
|
|
|
|
|
def _tool_section(name: str, report: SarifReport) -> dict[str, Any]:
|
|
"""Every finding of one tool that was not set aside."""
|
|
kept = _kept(report)
|
|
group: dict[str, Any] = {"title": f"True Issues ({len(kept)})", "cards": [_finding_card(one) for one in kept]}
|
|
|
|
if not kept:
|
|
group["note"] = f"Every finding {name} reported is listed under False Positives and Suppressed Findings."
|
|
|
|
return {"id": _anchor(name), "title": name, "new_page": True, "groups": [group]}
|
|
|
|
|
|
def _set_aside_section(tools: list[tuple[str, SarifReport]]) -> dict[str, Any] | None:
|
|
"""The false positives and the suppressed findings of every tool, at the end of the report."""
|
|
grouped = [(name, _set_aside(report)) for name, report in tools]
|
|
grouped = [(name, findings) for name, findings in grouped if findings]
|
|
|
|
if not grouped:
|
|
return None
|
|
|
|
groups: list[dict[str, Any]] = [
|
|
{
|
|
"title": "Overview",
|
|
"note": (
|
|
"The following findings were reported but set aside: classified as false positives, or "
|
|
"suppressed by the tool that reported them. Each entry carries the reason it was set aside."
|
|
),
|
|
"table": _set_aside_table(grouped),
|
|
}
|
|
]
|
|
groups += [
|
|
{"title": f"{name} ({len(findings)})", "cards": [_finding_card(one, set_aside=True) for one in findings]}
|
|
for name, findings in grouped
|
|
]
|
|
|
|
return {
|
|
"id": "set-aside",
|
|
"title": "False Positives and Suppressed Findings",
|
|
"new_page": True,
|
|
"groups": groups,
|
|
}
|
|
|
|
|
|
def _finding_card(finding: Finding, *, set_aside: bool = False) -> dict[str, Any]:
|
|
"""One SARIF finding."""
|
|
rows = []
|
|
|
|
# A finding a tool ruled out carries no severity worth stating.
|
|
if finding.level != "none":
|
|
rows.append(_row("Severity", _finding_severity(finding), kind="severity"))
|
|
|
|
# Not every finding points at a line: an area an agent ruled out names it in its message instead.
|
|
if finding.location is not None:
|
|
rows.append(_row("Location", _where(finding), kind="mono"))
|
|
|
|
rows += [
|
|
_row("Description", _text(finding.message)),
|
|
_row("Reported By", ", ".join(finding.tools) or None),
|
|
]
|
|
|
|
if set_aside:
|
|
reason = finding.polygraph_reasoning or "; ".join(
|
|
one.justification for one in finding.suppressions if one.justification
|
|
)
|
|
rows.append(_row("Verdict", finding.verdict_label or "Suppressed"))
|
|
rows.append(_row("Reason", reason, aside=True))
|
|
elif finding.verdict_label:
|
|
rows.append(_row("Evidence", finding.verdict_label))
|
|
|
|
return {"title": f"{_stated(finding.rule_id)} — {_stated(finding.rule_name)}", "rows": rows}
|
|
|
|
|
|
def _set_aside_table(grouped: list[tuple[str, list[Finding]]]) -> dict[str, Any]:
|
|
"""How many findings of each tool and rule were set aside."""
|
|
rows: list[dict[str, Any]] = []
|
|
|
|
for name, findings in grouped:
|
|
counts = Counter(one.rule_id or UNKNOWN for one in findings)
|
|
rows += [{"cells": [name, rule, count]} for rule, count in counts.most_common()]
|
|
|
|
rows.append({"cells": ["Total", "", sum(len(findings) for _, findings in grouped)], "total": True})
|
|
|
|
return {"columns": ["Tool", "Rule", "Set aside"], "rows": rows}
|
|
|
|
|
|
def _sca_section(sca: VdrReport) -> dict[str, Any]:
|
|
"""The dependency findings."""
|
|
return {
|
|
"id": "sca",
|
|
"title": "Dependency Vulnerabilities (SBOM / SCA)",
|
|
"new_page": True,
|
|
"groups": [
|
|
{
|
|
"title": f"True Issues by CVEs ({sca.total})",
|
|
"cards": [_vulnerability_card(one) for one in sca.vulnerabilities],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
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 _is_set_aside(finding: Finding) -> bool:
|
|
"""Whether a finding was classified as a false positive or suppressed by the tool."""
|
|
return finding.false_positive or bool(finding.suppressions)
|
|
|
|
|
|
def _kept(report: SarifReport) -> list[Finding]:
|
|
"""The findings that survived filtering."""
|
|
return [one for one in report.findings if not _is_set_aside(one)]
|
|
|
|
|
|
def _set_aside(report: SarifReport) -> list[Finding]:
|
|
"""The findings that were filtered out or suppressed."""
|
|
return [one for one in report.findings if _is_set_aside(one)]
|
|
|
|
|
|
def _text(message: str) -> str:
|
|
"""A message on one line. Paragraph breaks -- a title above its description -- would otherwise run together."""
|
|
return re.sub(r"\s*\n\s*\n\s*", " — ", message.strip())
|
|
|
|
|
|
def _anchor(title: str) -> str:
|
|
"""An id the table of contents can link to."""
|
|
return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "section"
|
|
|
|
|
|
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``, every location of it."""
|
|
if finding.location is None:
|
|
return None
|
|
|
|
return ", ".join(
|
|
f"{one.file}:{one.start_line}" if one.start_line else one.file
|
|
for one in [finding.location, *finding.extra_locations]
|
|
)
|
|
|
|
|
|
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}"
|