Initial commit. monorepo deps are copied in a dirty way to the vendor folder
This commit is contained in:
Florian Frank committed
commit
06593ac488
53 files changed
+7692
No files matched your search
@@ -0,0 +1 @@
|
||||
"""Top-level package for the CipherVerifier Django project."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Core Django app for CipherVerifier."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Service layer for pipeline orchestration."""
|
||||
@@ -0,0 +1,677 @@
|
||||
"""Summarising of SARIF reports for presentation in the UI.
|
||||
|
||||
The functions in here are deliberately free of any storage or Django dependency: they take an
|
||||
already parsed SARIF document and turn it into a plain summary that a template can render.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
TOP_LIST_SIZE = 8
|
||||
"""How many entries of a top list are shown before the "See all" button has to be used."""
|
||||
|
||||
LEVEL_ORDER = ["error", "warning", "note", "none"]
|
||||
|
||||
# Kept in sync with the --scan-chart-color-* palette in tailwind/input.css so the charts on the
|
||||
# scan page look like the ones on the dashboard.
|
||||
LEVEL_COLORS = {
|
||||
"error": "oklch(0.5 0.199 28.536)",
|
||||
"warning": "oklch(0.747 0.155 103.227)",
|
||||
"note": "oklch(0.451 0.251 265.29)",
|
||||
"none": "oklch(0.708 0 0)",
|
||||
}
|
||||
|
||||
# The verdicts polygraph can hand down, with the wording the UI uses for them.
|
||||
POLYGRAPH_VERDICT_LABELS = {
|
||||
"false_positive": "Likely false positive",
|
||||
"true_positive": "Likely true positive",
|
||||
"uncertain": "Uncertain",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Region:
|
||||
"""A span within a file. Carries no path of its own - the location around it holds that."""
|
||||
|
||||
start_line: int | None
|
||||
end_line: int | None = None
|
||||
start_column: int | None = None
|
||||
end_column: int | None = None
|
||||
|
||||
@property
|
||||
def display(self) -> str:
|
||||
"""The span as ``3``, ``3-5`` or ``3:13-26``, empty when it does not name a line."""
|
||||
if self.start_line is None:
|
||||
return ""
|
||||
# A span across lines reads as a line range; one within a single line adds its columns.
|
||||
if self.end_line and self.end_line != self.start_line:
|
||||
return f"{self.start_line}-{self.end_line}"
|
||||
if self.start_column is None:
|
||||
return str(self.start_line)
|
||||
columns = str(self.start_column)
|
||||
if self.end_column and self.end_column != self.start_column:
|
||||
columns = f"{columns}-{self.end_column}"
|
||||
return f"{self.start_line}:{columns}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Occurrence:
|
||||
"""A single tool hit that was merged into a finding by the sarifcombiner."""
|
||||
|
||||
found_by: str
|
||||
tool_identifier: str
|
||||
message: str
|
||||
start_line: int | None
|
||||
end_line: int | None = None
|
||||
start_column: int | None = None
|
||||
end_column: int | None = None
|
||||
related_positions: list[Region] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def region(self) -> Region:
|
||||
"""The span this tool reported the finding at."""
|
||||
return Region(self.start_line, self.end_line, self.start_column, self.end_column)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Location:
|
||||
"""A source location a finding points at."""
|
||||
|
||||
file: str
|
||||
start_line: int | None
|
||||
end_line: int | None
|
||||
start_column: int | None = None
|
||||
end_column: int | None = None
|
||||
|
||||
@property
|
||||
def region(self) -> Region:
|
||||
"""The span within the file."""
|
||||
return Region(self.start_line, self.end_line, self.start_column, self.end_column)
|
||||
|
||||
@property
|
||||
def display(self) -> str:
|
||||
"""The location as ``app/views.py:3:13-26``, degrading to just the path without a line."""
|
||||
region = self.region.display
|
||||
return f"{self.file}:{region}" if region else self.file
|
||||
|
||||
|
||||
@dataclass
|
||||
class Suppression:
|
||||
"""A reason the finding was suppressed. Polygraph adds one for every false positive it finds."""
|
||||
|
||||
kind: str
|
||||
justification: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Finding:
|
||||
"""A single SARIF result, flattened into the fields the template needs."""
|
||||
|
||||
rule_id: str
|
||||
rule_name: str
|
||||
rule_description: str
|
||||
level: str
|
||||
message: str
|
||||
location: Location | None
|
||||
extra_locations: list[Location]
|
||||
tools: list[str]
|
||||
occurrences: list[Occurrence]
|
||||
rank: int | None
|
||||
polygraph_verdict: str | None = None
|
||||
polygraph_reasoning: str | None = None
|
||||
polygraph_certainty: str = ""
|
||||
polygraph_priority: str = ""
|
||||
polygraph_prompt_hash: str = ""
|
||||
rule_full_description: str = ""
|
||||
rule_help: str = ""
|
||||
rule_help_uri: str = ""
|
||||
rule_tags: list[str] = field(default_factory=list)
|
||||
rule_precision: str = ""
|
||||
rule_security_severity: str = ""
|
||||
rule_default_level: str = ""
|
||||
suppressions: list[Suppression] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def tool_count(self) -> int:
|
||||
"""Number of distinct tools that reported this finding."""
|
||||
return len(self.tools)
|
||||
|
||||
@property
|
||||
def downgraded(self) -> bool:
|
||||
"""Whether this result is reported below the severity its rule defaults to.
|
||||
|
||||
Tools do this per hit, so the gap is worth showing: a rule that defaults to ``error`` and
|
||||
lands as a ``note`` says the tool itself was unsure about this particular match. The
|
||||
reverse (a hit reported *above* the rule default) is not flagged here.
|
||||
"""
|
||||
if self.rule_default_level not in LEVEL_ORDER:
|
||||
return False
|
||||
return LEVEL_ORDER.index(self.level) > LEVEL_ORDER.index(self.rule_default_level)
|
||||
|
||||
@property
|
||||
def false_positive(self) -> bool:
|
||||
"""Whether polygraph classified this finding as a likely false positive."""
|
||||
return self.polygraph_verdict == "false_positive"
|
||||
|
||||
@property
|
||||
def verdict_label(self) -> str:
|
||||
"""The polygraph verdict in words, empty when the finding carries no known verdict."""
|
||||
return POLYGRAPH_VERDICT_LABELS.get(self.polygraph_verdict or "", "")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Stat:
|
||||
"""A labelled count with its share of the largest count in the same list, in percent."""
|
||||
|
||||
label: str
|
||||
sublabel: str
|
||||
count: int
|
||||
percent: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class SarifReport:
|
||||
"""Everything the scan page shows about one SARIF report."""
|
||||
|
||||
total: int
|
||||
findings: list[Finding]
|
||||
level_counts: list[Stat]
|
||||
tool_counts: list[Stat]
|
||||
top_rules: list[Stat]
|
||||
top_files: list[Stat]
|
||||
file_count: int
|
||||
rule_count: int
|
||||
multi_tool_count: int
|
||||
false_positive_count: int
|
||||
ranked: bool
|
||||
chart_data: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def levels(self) -> dict[str, int]:
|
||||
"""Finding count per severity level, including the levels without any finding."""
|
||||
counts = dict.fromkeys(LEVEL_ORDER, 0)
|
||||
for stat in self.level_counts:
|
||||
counts[stat.label] = stat.count
|
||||
return counts
|
||||
|
||||
@property
|
||||
def top_list_size(self) -> int:
|
||||
"""How many entries of ``top_rules`` and ``top_files`` the page shows without unfolding."""
|
||||
return TOP_LIST_SIZE
|
||||
|
||||
|
||||
def _rules_by_id(runs: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
||||
"""Collect the rule definitions of every run, keyed by rule id."""
|
||||
rules: dict[str, dict[str, Any]] = {}
|
||||
for run in runs:
|
||||
for rule in run.get("tool", {}).get("driver", {}).get("rules", []):
|
||||
if isinstance(rule, dict) and "id" in rule:
|
||||
rules.setdefault(rule["id"], rule)
|
||||
return rules
|
||||
|
||||
|
||||
def _driver_name(run: dict[str, Any]) -> str:
|
||||
"""Return the display name of the tool that produced a run."""
|
||||
driver = run.get("tool", {}).get("driver", {})
|
||||
return str(driver.get("name") or driver.get("fullName") or "unknown")
|
||||
|
||||
|
||||
def _level_of(result: dict[str, Any], rule: dict[str, Any]) -> str:
|
||||
"""Return the severity of a result.
|
||||
|
||||
Some tools (semgrep for instance) omit ``level`` on the result and only declare it on the rule,
|
||||
so the rule's default configuration is used as a fallback.
|
||||
|
||||
Args:
|
||||
result: A SARIF result.
|
||||
rule: The rule definition belonging to the result, or an empty dict.
|
||||
|
||||
Returns:
|
||||
One of the levels in ``LEVEL_ORDER``.
|
||||
"""
|
||||
level = result.get("level") or rule.get("defaultConfiguration", {}).get("level")
|
||||
return level if level in LEVEL_ORDER else "none"
|
||||
|
||||
|
||||
def _region_of(region: dict[str, Any]) -> Region:
|
||||
"""Read a SARIF region, or a sarifcombiner position - both spell the keys the same way."""
|
||||
return Region(
|
||||
start_line=region.get("startLine"),
|
||||
end_line=region.get("endLine"),
|
||||
start_column=region.get("startColumn"),
|
||||
end_column=region.get("endColumn"),
|
||||
)
|
||||
|
||||
|
||||
def _locations_of(result: dict[str, Any]) -> list[Location]:
|
||||
"""Extract all physical locations of a result."""
|
||||
locations = []
|
||||
for location in result.get("locations", []):
|
||||
physical = location.get("physicalLocation", {})
|
||||
uri = physical.get("artifactLocation", {}).get("uri")
|
||||
if not uri:
|
||||
continue
|
||||
region = _region_of(physical.get("region", {}))
|
||||
locations.append(
|
||||
Location(
|
||||
file=uri,
|
||||
start_line=region.start_line,
|
||||
end_line=region.end_line,
|
||||
start_column=region.start_column,
|
||||
end_column=region.end_column,
|
||||
)
|
||||
)
|
||||
return locations
|
||||
|
||||
|
||||
def _occurrences_of(result: dict[str, Any]) -> list[Occurrence]:
|
||||
"""Extract the per-tool occurrences the sarifcombiner merged into a result."""
|
||||
occurrences = []
|
||||
for occurrence in result.get("properties", {}).get("sarifgrouper", {}).get("occurrences", []):
|
||||
found_by = occurrence.get("foundBy")
|
||||
if not found_by:
|
||||
continue
|
||||
position = _region_of(occurrence.get("position", {}) or {})
|
||||
related = [_region_of(related) for related in occurrence.get("relatedPositions", []) or []]
|
||||
occurrences.append(
|
||||
Occurrence(
|
||||
found_by=found_by,
|
||||
tool_identifier=occurrence.get("toolIdentifier", ""),
|
||||
message=occurrence.get("message", ""),
|
||||
start_line=position.start_line,
|
||||
end_line=position.end_line,
|
||||
start_column=position.start_column,
|
||||
end_column=position.end_column,
|
||||
related_positions=[region for region in related if region.start_line is not None],
|
||||
)
|
||||
)
|
||||
return occurrences
|
||||
|
||||
|
||||
def _suppressions_of(result: dict[str, Any]) -> list[Suppression]:
|
||||
"""Extract the suppressions of a result. Polygraph adds one for each false positive it finds."""
|
||||
suppressions = []
|
||||
for suppression in result.get("suppressions", []) or []:
|
||||
if not isinstance(suppression, dict):
|
||||
continue
|
||||
suppressions.append(
|
||||
Suppression(
|
||||
kind=str(suppression.get("kind") or ""),
|
||||
justification=str(suppression.get("justification") or ""),
|
||||
)
|
||||
)
|
||||
return suppressions
|
||||
|
||||
|
||||
def _rank_of(result: dict[str, Any]) -> int | None:
|
||||
"""Return the prism rank of a result, if the report was ranked."""
|
||||
rank = result.get("properties", {}).get("prism", {}).get("rank")
|
||||
if isinstance(rank, bool) or rank is None:
|
||||
return None
|
||||
if isinstance(rank, int | float):
|
||||
return int(rank)
|
||||
if isinstance(rank, str) and rank.isdigit():
|
||||
return int(rank)
|
||||
return None
|
||||
|
||||
|
||||
POLYGRAPH_VERDICTS = set(POLYGRAPH_VERDICT_LABELS)
|
||||
|
||||
|
||||
EMPTY_POLYGRAPH: dict[str, Any] = {"verdict": None, "reasoning": None, "certainty": "", "priority": "", "hash": ""}
|
||||
|
||||
|
||||
def _polygraph_of(result: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Return what polygraph concluded about a result, if it was classified.
|
||||
|
||||
Polygraph annotates each classified SARIF result with a ``polygraph`` property block. Findings
|
||||
that were not classified (polygraph disabled, below the severity threshold, or errored) carry no
|
||||
such block, so an empty verdict is returned and the finding reads as "not classified".
|
||||
|
||||
Args:
|
||||
result: A SARIF result.
|
||||
|
||||
Returns:
|
||||
A dict with the ``verdict`` (one of ``POLYGRAPH_VERDICTS`` or ``None``), the ``reasoning``,
|
||||
and the ``certainty``, ``priority`` and prompt ``hash`` of the classification. Certainty and
|
||||
priority are lowercased, so the templates can capitalise them like they do severities.
|
||||
"""
|
||||
polygraph = result.get("properties", {}).get("polygraph")
|
||||
if not isinstance(polygraph, dict):
|
||||
return dict(EMPTY_POLYGRAPH)
|
||||
verdict = str(polygraph.get("verdict", "")).lower()
|
||||
if verdict not in POLYGRAPH_VERDICTS:
|
||||
# Tolerate an older shape that only carries the boolean flag.
|
||||
verdict = "false_positive" if polygraph.get("isFalsePositive") else ""
|
||||
reasoning = polygraph.get("reasoning")
|
||||
return {
|
||||
"verdict": verdict or None,
|
||||
"reasoning": str(reasoning) if reasoning else None,
|
||||
"certainty": str(polygraph.get("certainty") or "").lower(),
|
||||
"priority": str(polygraph.get("priority") or "").lower(),
|
||||
"hash": str(polygraph.get("promptHash") or ""),
|
||||
}
|
||||
|
||||
|
||||
def _rule_metadata(rule: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Pull everything a rule definition says about itself.
|
||||
|
||||
Which of these a rule carries is entirely up to the tool that declared it: the sarifcombiner
|
||||
keeps descriptions and tags, while raw tool reports (rayfusion's, for instance) also come with
|
||||
a ``helpUri`` and a precision. Anything absent comes back empty and the template leaves it out.
|
||||
|
||||
Args:
|
||||
rule: A SARIF rule definition, or an empty dict when the rule was not declared.
|
||||
|
||||
Returns:
|
||||
The rule's descriptions, help text and link, tags, precision, security severity and the
|
||||
severity it defaults to.
|
||||
"""
|
||||
properties = rule.get("properties") or {}
|
||||
tags = properties.get("tags")
|
||||
security_severity = properties.get("security-severity", properties.get("securitySeverity"))
|
||||
return {
|
||||
"short_description": rule.get("shortDescription", {}).get("text") or "",
|
||||
"full_description": rule.get("fullDescription", {}).get("text") or "",
|
||||
"help": (rule.get("help") or {}).get("text") or "",
|
||||
"help_uri": str(rule.get("helpUri") or ""),
|
||||
"tags": [str(tag) for tag in tags] if isinstance(tags, list) else [],
|
||||
"precision": str(properties.get("precision") or ""),
|
||||
"security_severity": str(security_severity) if security_severity is not None else "",
|
||||
"default_level": str(rule.get("defaultConfiguration", {}).get("level") or ""),
|
||||
}
|
||||
|
||||
|
||||
def _build_finding(result: dict[str, Any], run: dict[str, Any], rules: dict[str, dict[str, Any]]) -> Finding:
|
||||
"""Flatten a single SARIF result into a ``Finding``.
|
||||
|
||||
Args:
|
||||
result: The SARIF result to flatten.
|
||||
run: The run the result belongs to, used as fallback for the tool name.
|
||||
rules: All rule definitions of the report, keyed by rule id.
|
||||
|
||||
Returns:
|
||||
The flattened finding.
|
||||
"""
|
||||
rule_id = str(result.get("ruleId", ""))
|
||||
rule = rules.get(rule_id, {})
|
||||
meta = _rule_metadata(rule)
|
||||
occurrences = _occurrences_of(result)
|
||||
locations = _locations_of(result)
|
||||
polygraph = _polygraph_of(result)
|
||||
|
||||
# The full description repeats the short one when a tool declares only one of the two, so it is
|
||||
# dropped rather than shown twice underneath it.
|
||||
description = meta["short_description"] or meta["full_description"]
|
||||
full_description = meta["full_description"] if meta["full_description"] != description else ""
|
||||
|
||||
return Finding(
|
||||
rule_id=rule_id,
|
||||
rule_name=rule.get("name", ""),
|
||||
rule_description=description,
|
||||
level=_level_of(result, rule),
|
||||
message=result.get("message", {}).get("text", ""),
|
||||
location=locations[0] if locations else None,
|
||||
extra_locations=locations[1:],
|
||||
tools=sorted({occurrence.found_by for occurrence in occurrences}) or [_driver_name(run)],
|
||||
occurrences=occurrences,
|
||||
rank=_rank_of(result),
|
||||
polygraph_verdict=polygraph["verdict"],
|
||||
polygraph_reasoning=polygraph["reasoning"],
|
||||
polygraph_certainty=polygraph["certainty"],
|
||||
polygraph_priority=polygraph["priority"],
|
||||
polygraph_prompt_hash=polygraph["hash"],
|
||||
rule_full_description=full_description,
|
||||
rule_help=meta["help"] if meta["help"] not in (description, full_description) else "",
|
||||
rule_help_uri=meta["help_uri"],
|
||||
rule_tags=meta["tags"],
|
||||
rule_precision=meta["precision"],
|
||||
rule_security_severity=meta["security_severity"],
|
||||
rule_default_level=meta["default_level"],
|
||||
suppressions=_suppressions_of(result),
|
||||
)
|
||||
|
||||
|
||||
def _sort_findings(findings: list[Finding], ranked: bool) -> list[Finding]:
|
||||
"""Sort findings, keeping the ones polygraph flagged as false positives at the bottom.
|
||||
|
||||
The likely false positives sink below every real, uncertain or unclassified finding; within
|
||||
each of those two groups the usual order is kept - by rank when the report is ranked, by
|
||||
severity and location otherwise.
|
||||
"""
|
||||
if ranked:
|
||||
return sorted(findings, key=lambda f: (f.false_positive, f.rank is None, f.rank or 0))
|
||||
return sorted(
|
||||
findings,
|
||||
key=lambda f: (
|
||||
f.false_positive,
|
||||
LEVEL_ORDER.index(f.level),
|
||||
f.location.file if f.location else "",
|
||||
f.location.start_line or 0 if f.location else 0,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _to_stats(counts: list[tuple[str, int]], sublabels: dict[str, str] | None = None) -> list[Stat]:
|
||||
"""Turn ``(label, count)`` pairs into ``Stat`` entries with a percentage of the largest count."""
|
||||
if not counts:
|
||||
return []
|
||||
largest = max(count for _, count in counts)
|
||||
return [
|
||||
Stat(
|
||||
label=label,
|
||||
sublabel=(sublabels or {}).get(label, ""),
|
||||
count=count,
|
||||
percent=round(count / largest * 100) if largest else 0,
|
||||
)
|
||||
for label, count in counts
|
||||
]
|
||||
|
||||
|
||||
def _level_stats(findings: list[Finding]) -> list[Stat]:
|
||||
"""Count findings per severity level, in severity order and without empty buckets."""
|
||||
counter = Counter(finding.level for finding in findings)
|
||||
return _to_stats([(level, counter[level]) for level in LEVEL_ORDER if counter[level]])
|
||||
|
||||
|
||||
def _tool_stats(findings: list[Finding]) -> list[Stat]:
|
||||
"""Count findings per reporting tool, most findings first."""
|
||||
counter: Counter[str] = Counter()
|
||||
for finding in findings:
|
||||
counter.update(finding.tools)
|
||||
return _to_stats(counter.most_common())
|
||||
|
||||
|
||||
def _rule_stats(findings: list[Finding]) -> list[Stat]:
|
||||
"""Rank every triggered rule by how often it was triggered, most triggered first.
|
||||
|
||||
The scan page renders the whole list: it shows the first ``TOP_LIST_SIZE`` entries and reveals
|
||||
the rest behind a "See all" button.
|
||||
"""
|
||||
counter = Counter(finding.rule_id for finding in findings if finding.rule_id)
|
||||
names = {finding.rule_id: finding.rule_name for finding in findings}
|
||||
return _to_stats(counter.most_common(), sublabels=names)
|
||||
|
||||
|
||||
def _file_stats(findings: list[Finding]) -> list[Stat]:
|
||||
"""Rank every affected file by its number of findings, most findings first."""
|
||||
counter = Counter(finding.location.file for finding in findings if finding.location)
|
||||
return _to_stats(counter.most_common())
|
||||
|
||||
|
||||
def _chart_data(level_counts: list[Stat], tool_counts: list[Stat]) -> dict[str, Any]:
|
||||
"""Build the data blob the Chart.js charts on the scan page are fed with."""
|
||||
return {
|
||||
"severity": {
|
||||
"labels": [stat.label.capitalize() for stat in level_counts],
|
||||
"values": [stat.count for stat in level_counts],
|
||||
"colors": [LEVEL_COLORS[stat.label] for stat in level_counts],
|
||||
},
|
||||
"tools": {
|
||||
"labels": [stat.label for stat in tool_counts],
|
||||
"values": [stat.count for stat in tool_counts],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _build_report(findings: list[Finding], ranked: bool) -> SarifReport:
|
||||
"""Sort a list of findings and derive every statistic the templates render from it."""
|
||||
findings = _sort_findings(findings, ranked)
|
||||
|
||||
level_counts = _level_stats(findings)
|
||||
tool_counts = _tool_stats(findings)
|
||||
|
||||
return SarifReport(
|
||||
total=len(findings),
|
||||
findings=findings,
|
||||
level_counts=level_counts,
|
||||
tool_counts=tool_counts,
|
||||
top_rules=_rule_stats(findings),
|
||||
top_files=_file_stats(findings),
|
||||
file_count=len({finding.location.file for finding in findings if finding.location}),
|
||||
rule_count=len({finding.rule_id for finding in findings if finding.rule_id}),
|
||||
multi_tool_count=sum(1 for finding in findings if finding.tool_count > 1),
|
||||
false_positive_count=sum(1 for finding in findings if finding.false_positive),
|
||||
ranked=ranked,
|
||||
chart_data=_chart_data(level_counts, tool_counts),
|
||||
)
|
||||
|
||||
|
||||
def summarize_sarif(sarif_data: dict[str, Any]) -> SarifReport:
|
||||
"""Summarise a parsed SARIF document for rendering on the scan page.
|
||||
|
||||
All runs of the document are merged, so reports that keep one run per tool are summarised just
|
||||
as well as the combined report the pipeline usually produces. Every finding of the report is
|
||||
included; the scan page renders the complete list.
|
||||
|
||||
Args:
|
||||
sarif_data: A parsed SARIF document.
|
||||
|
||||
Returns:
|
||||
The summarised report.
|
||||
"""
|
||||
runs = [run for run in sarif_data.get("runs", []) if isinstance(run, dict)]
|
||||
rules = _rules_by_id(runs)
|
||||
|
||||
findings = [_build_finding(result, run, rules) for run in runs for result in run.get("results", [])]
|
||||
|
||||
return _build_report(findings, ranked=any(finding.rank is not None for finding in findings))
|
||||
|
||||
|
||||
@dataclass
|
||||
class FindingsSummary:
|
||||
"""The numbers of one or more SARIF reports, in a form that can be stored and added up.
|
||||
|
||||
The listing pages only show counts, never individual findings, so they work off this instead of
|
||||
the full report: it is small enough to cache per scan and can be merged across repositories
|
||||
without having the findings themselves at hand.
|
||||
"""
|
||||
|
||||
total: int = 0
|
||||
file_count: int = 0
|
||||
multi_tool_count: int = 0
|
||||
false_positive_count: int = 0
|
||||
level_totals: dict[str, int] = field(default_factory=dict)
|
||||
rule_totals: dict[str, int] = field(default_factory=dict)
|
||||
rule_names: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def levels(self) -> dict[str, int]:
|
||||
"""Finding count per severity level, including the levels without any finding."""
|
||||
counts = dict.fromkeys(LEVEL_ORDER, 0)
|
||||
for level, count in self.level_totals.items():
|
||||
if level in counts:
|
||||
counts[level] = count
|
||||
return counts
|
||||
|
||||
@property
|
||||
def level_counts(self) -> list[Stat]:
|
||||
"""Findings per severity level, in severity order and without empty buckets."""
|
||||
levels = self.levels
|
||||
return _to_stats([(level, levels[level]) for level in LEVEL_ORDER if levels[level]])
|
||||
|
||||
@property
|
||||
def rule_count(self) -> int:
|
||||
"""Number of distinct rules that were triggered."""
|
||||
return len(self.rule_totals)
|
||||
|
||||
@property
|
||||
def top_rules(self) -> list[Stat]:
|
||||
"""The most frequently triggered rules."""
|
||||
return _to_stats(Counter(self.rule_totals).most_common(TOP_LIST_SIZE), sublabels=self.rule_names)
|
||||
|
||||
@classmethod
|
||||
def of(cls, report: SarifReport) -> FindingsSummary:
|
||||
"""Reduce a full report to the numbers the listing pages need."""
|
||||
return cls(
|
||||
total=report.total,
|
||||
file_count=report.file_count,
|
||||
multi_tool_count=report.multi_tool_count,
|
||||
false_positive_count=report.false_positive_count,
|
||||
level_totals=report.levels,
|
||||
rule_totals=dict(Counter(finding.rule_id for finding in report.findings if finding.rule_id)),
|
||||
rule_names={
|
||||
finding.rule_id: finding.rule_name
|
||||
for finding in report.findings
|
||||
if finding.rule_id and finding.rule_name
|
||||
},
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialise the summary so it can be stored as JSON."""
|
||||
return {
|
||||
"total": self.total,
|
||||
"file_count": self.file_count,
|
||||
"multi_tool_count": self.multi_tool_count,
|
||||
"false_positive_count": self.false_positive_count,
|
||||
"level_totals": self.level_totals,
|
||||
"rule_totals": self.rule_totals,
|
||||
"rule_names": self.rule_names,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> FindingsSummary:
|
||||
"""Rebuild a summary from stored JSON, tolerating entries that miss a key."""
|
||||
return cls(
|
||||
total=int(data.get("total", 0)),
|
||||
file_count=int(data.get("file_count", 0)),
|
||||
multi_tool_count=int(data.get("multi_tool_count", 0)),
|
||||
false_positive_count=int(data.get("false_positive_count", 0)),
|
||||
level_totals=dict(data.get("level_totals") or {}),
|
||||
rule_totals=dict(data.get("rule_totals") or {}),
|
||||
rule_names=dict(data.get("rule_names") or {}),
|
||||
)
|
||||
|
||||
|
||||
def merge_summaries(summaries: list[FindingsSummary]) -> FindingsSummary:
|
||||
"""Add up the summaries of several scans.
|
||||
|
||||
Files are counted per scan, so the same path in two repositories counts twice - which is what
|
||||
the listing pages want, as those are two different files.
|
||||
|
||||
Args:
|
||||
summaries: The summaries to add up.
|
||||
|
||||
Returns:
|
||||
One summary covering all of them.
|
||||
"""
|
||||
levels: Counter[str] = Counter()
|
||||
rules: Counter[str] = Counter()
|
||||
names: dict[str, str] = {}
|
||||
for summary in summaries:
|
||||
levels.update(summary.level_totals)
|
||||
rules.update(summary.rule_totals)
|
||||
names.update(summary.rule_names)
|
||||
|
||||
return FindingsSummary(
|
||||
total=sum(summary.total for summary in summaries),
|
||||
file_count=sum(summary.file_count for summary in summaries),
|
||||
multi_tool_count=sum(summary.multi_tool_count for summary in summaries),
|
||||
false_positive_count=sum(summary.false_positive_count for summary in summaries),
|
||||
level_totals=dict(levels),
|
||||
rule_totals=dict(rules),
|
||||
rule_names=names,
|
||||
)
|
||||
@@ -0,0 +1,813 @@
|
||||
"""Summarising of vulnerability reports for presentation in the UI.
|
||||
|
||||
Two shapes end up in the ``vdr.json`` artifact of a scan: the Finding Packaging Format that
|
||||
Dependency-Track's findings export produces, and a plain CycloneDX VDR document. Both are accepted
|
||||
here and reduced to the same summary.
|
||||
|
||||
The functions in here are deliberately free of any storage or Django dependency: they take an
|
||||
already parsed document and turn it into a plain summary that a template can render.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections import Counter
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from cipherverifier.core.services.sarif import TOP_LIST_SIZE, Stat
|
||||
from persistent_storage.utils.cvss_utils import IMPACT_KEYS, describe, parse_vector
|
||||
|
||||
# CycloneDX severity values, most severe first. ``unknown`` covers reports that carry no rating.
|
||||
SEVERITY_ORDER = ["critical", "high", "medium", "low", "info", "unknown"]
|
||||
|
||||
# Higher rank means more severe, so the highest rated severity of a vulnerability wins.
|
||||
SEVERITY_RANK: dict[str, int] = {
|
||||
"critical": 5,
|
||||
"high": 4,
|
||||
"medium": 3,
|
||||
"low": 2,
|
||||
"info": 1,
|
||||
"unknown": 0,
|
||||
}
|
||||
|
||||
# Kept as OKLCH so the scan page charts match the style of the SARIF charts.
|
||||
SEVERITY_COLORS: dict[str, str] = {
|
||||
"critical": "oklch(0.45 0.22 27)",
|
||||
"high": "oklch(0.62 0.20 38)",
|
||||
"medium": "oklch(0.75 0.15 80)",
|
||||
"low": "oklch(0.65 0.12 250)",
|
||||
"info": "oklch(0.708 0 0)",
|
||||
"unknown": "oklch(0.708 0 0)",
|
||||
}
|
||||
|
||||
|
||||
# How CycloneDX names each CVSS generation in a rating's ``method``, by which one is newer.
|
||||
CYCLONEDX_METHODS = {"cvssv2": 2, "cvssv3": 3, "cvssv31": 3, "cvssv4": 4}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Component:
|
||||
"""A dependency a vulnerability affects.
|
||||
|
||||
Attributes:
|
||||
name: What the package is called.
|
||||
version: The version this scan depends on, empty when the report named none.
|
||||
latest_version: The newest version the reporting tool knew of. What makes
|
||||
a finding answerable -- a newer version existing is not proof that it
|
||||
fixes this advisory, only that there is somewhere to upgrade to.
|
||||
"""
|
||||
|
||||
name: str
|
||||
version: str
|
||||
latest_version: str = ""
|
||||
|
||||
@property
|
||||
def is_outdated(self) -> bool:
|
||||
"""Whether the package has been released again since the version in use."""
|
||||
return bool(self.latest_version) and self.latest_version != self.version
|
||||
|
||||
|
||||
@dataclass
|
||||
class VulnerabilityMetrics:
|
||||
"""What one advisory's CVSS vector and scoring say, in the words a reader uses.
|
||||
|
||||
Attributes:
|
||||
attack_vector: Where an attacker has to be.
|
||||
attack_complexity: Whether anything outside their control has to line up.
|
||||
privileges_required: What account they need. v2 states how often they log
|
||||
in instead, which is the same question.
|
||||
user_interaction: Who they have to fool.
|
||||
confidentiality: What they get to read.
|
||||
integrity: What they get to change.
|
||||
availability: How far they can take the service down.
|
||||
epss: The probability of exploitation in the next 30 days, 0 to 1, or
|
||||
``None`` where nobody scored it -- which is not a probability of zero.
|
||||
epss_percentile: Where that probability ranks among everything scored.
|
||||
published_days_ago: How long the advisory has been public.
|
||||
"""
|
||||
|
||||
attack_vector: str = ""
|
||||
attack_complexity: str = ""
|
||||
privileges_required: str = ""
|
||||
user_interaction: str = ""
|
||||
confidentiality: str = ""
|
||||
integrity: str = ""
|
||||
availability: str = ""
|
||||
epss: float | None = None
|
||||
epss_percentile: float | None = None
|
||||
published_days_ago: int | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Vulnerability:
|
||||
"""A single CycloneDX vulnerability, flattened into the fields the template needs."""
|
||||
|
||||
id: str
|
||||
severity: str
|
||||
description: str
|
||||
cwes: list[str]
|
||||
components: list[Component]
|
||||
score: float | None
|
||||
recommendation: str
|
||||
url: str
|
||||
metrics: VulnerabilityMetrics = field(default_factory=VulnerabilityMetrics)
|
||||
cwe_names: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def cwes_json(self) -> str:
|
||||
"""The weakness classes as one JSON array of identifier and name.
|
||||
|
||||
One attribute rather than two parallel pipe-joined lists, which would have to be kept in
|
||||
step by whoever writes the template and would silently mispair if they ever were not.
|
||||
A class the report named without describing carries an empty name, and the panel shows the
|
||||
identifier alone rather than inventing a description for it.
|
||||
"""
|
||||
return json.dumps(
|
||||
[{"id": one, "name": self.cwe_names.get(one, "")} for one in self.cwes],
|
||||
separators=(",", ":"),
|
||||
)
|
||||
|
||||
@property
|
||||
def fix_available(self) -> bool:
|
||||
"""Whether any affected package has been released again since."""
|
||||
return any(component.is_outdated for component in self.components)
|
||||
|
||||
@property
|
||||
def metrics_json(self) -> str:
|
||||
"""The metrics as one JSON object, for the row to carry into the panel.
|
||||
|
||||
One attribute rather than ten: the panel needs all of them or none, and a single blob
|
||||
cannot end up half-updated by a template that forgot a field. Keyed the way the script
|
||||
reads them, which is the same shape its sample values have -- so the panel does not know
|
||||
or care which of the two it was handed.
|
||||
"""
|
||||
return json.dumps(
|
||||
{
|
||||
"score": self.score,
|
||||
"band": self.severity.capitalize(),
|
||||
"epss": self.metrics.epss,
|
||||
"epssPercentile": self.metrics.epss_percentile,
|
||||
"confidentiality": self.metrics.confidentiality,
|
||||
"integrity": self.metrics.integrity,
|
||||
"availability": self.metrics.availability,
|
||||
"attackVector": self.metrics.attack_vector,
|
||||
"attackComplexity": self.metrics.attack_complexity,
|
||||
"privilegesRequired": self.metrics.privileges_required,
|
||||
"userInteraction": self.metrics.user_interaction,
|
||||
"publishedDaysAgo": self.metrics.published_days_ago,
|
||||
"fixAvailable": self.fix_available,
|
||||
},
|
||||
separators=(",", ":"),
|
||||
)
|
||||
|
||||
@property
|
||||
def component_names(self) -> list[str]:
|
||||
"""Display names of the affected components, name plus version when the version is known."""
|
||||
return [
|
||||
f"{component.name}@{component.version}" if component.version else component.name
|
||||
for component in self.components
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class VdrReport:
|
||||
"""Everything the scan page shows about one CycloneDX VDR report."""
|
||||
|
||||
total: int
|
||||
vulnerabilities: list[Vulnerability]
|
||||
severity_counts: list[Stat]
|
||||
top_components: list[Stat]
|
||||
component_count: int
|
||||
chart_data: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def critical_high_count(self) -> int:
|
||||
"""Number of vulnerabilities rated critical or high."""
|
||||
return sum(1 for vuln in self.vulnerabilities if vuln.severity in ("critical", "high"))
|
||||
|
||||
|
||||
def _components_by_ref(data: dict[str, Any]) -> dict[str, Component]:
|
||||
"""Map the bom-ref of every component to a ``Component``, for resolving ``affects``.
|
||||
|
||||
Args:
|
||||
data: A parsed CycloneDX VDR document.
|
||||
|
||||
Returns:
|
||||
The components keyed by their bom-ref.
|
||||
"""
|
||||
components: dict[str, Component] = {}
|
||||
for component in data.get("components", []):
|
||||
if not isinstance(component, dict):
|
||||
continue
|
||||
bom_ref = component.get("bom-ref") or component.get("purl")
|
||||
if not bom_ref:
|
||||
continue
|
||||
components[bom_ref] = Component(
|
||||
name=str(component.get("name") or bom_ref),
|
||||
version=str(component.get("version") or ""),
|
||||
)
|
||||
return components
|
||||
|
||||
|
||||
def _severity_of(vulnerability: dict[str, Any]) -> str:
|
||||
"""Return the most severe rating of a vulnerability, ``unknown`` when none is given.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
|
||||
Returns:
|
||||
One of the severities in ``SEVERITY_ORDER``.
|
||||
"""
|
||||
best = "unknown"
|
||||
for rating in vulnerability.get("ratings", []):
|
||||
if not isinstance(rating, dict):
|
||||
continue
|
||||
severity = rating.get("severity")
|
||||
if isinstance(severity, str) and severity in SEVERITY_RANK and SEVERITY_RANK[severity] > SEVERITY_RANK[best]:
|
||||
best = severity
|
||||
return best
|
||||
|
||||
|
||||
def _score_of(vulnerability: dict[str, Any]) -> float | None:
|
||||
"""Return the highest CVSS score carried by any rating of a vulnerability.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
|
||||
Returns:
|
||||
The highest score, or ``None`` when no rating has a numeric score.
|
||||
"""
|
||||
scores = [
|
||||
float(rating["score"])
|
||||
for rating in vulnerability.get("ratings", [])
|
||||
if isinstance(rating, dict) and isinstance(rating.get("score"), int | float)
|
||||
]
|
||||
return max(scores) if scores else None
|
||||
|
||||
|
||||
def _cwes_of(vulnerability: dict[str, Any]) -> list[str]:
|
||||
"""Return the CWE ids of a vulnerability as ``CWE-<n>`` strings.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
|
||||
Returns:
|
||||
The CWE ids, normalised to the ``CWE-<n>`` form.
|
||||
"""
|
||||
cwes: list[str] = []
|
||||
for value in vulnerability.get("cwes", []):
|
||||
if isinstance(value, int):
|
||||
cwes.append(f"CWE-{value}")
|
||||
elif isinstance(value, str) and value:
|
||||
cwes.append(value if value.startswith("CWE-") else f"CWE-{value}")
|
||||
return cwes
|
||||
|
||||
|
||||
def _components_of(vulnerability: dict[str, Any], components: dict[str, Component]) -> list[Component]:
|
||||
"""Resolve the affected components of a vulnerability via their bom-ref.
|
||||
|
||||
Affects whose reference is unknown are still listed, using the reference itself as the name so
|
||||
the vulnerability is never left without a target.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
components: All components of the document, keyed by bom-ref.
|
||||
|
||||
Returns:
|
||||
The resolved components the vulnerability affects.
|
||||
"""
|
||||
resolved: list[Component] = []
|
||||
for affect in vulnerability.get("affects", []):
|
||||
ref = affect.get("ref") if isinstance(affect, dict) else None
|
||||
if not ref:
|
||||
continue
|
||||
resolved.append(components.get(ref, Component(name=str(ref), version="")))
|
||||
return resolved
|
||||
|
||||
|
||||
def _url_of(vulnerability: dict[str, Any]) -> str:
|
||||
"""Return a reference url for a vulnerability: its source url or the first advisory.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
|
||||
Returns:
|
||||
A url, or an empty string when the vulnerability carries none.
|
||||
"""
|
||||
source_url = vulnerability.get("source", {})
|
||||
if isinstance(source_url, dict) and source_url.get("url"):
|
||||
return str(source_url["url"])
|
||||
for advisory in vulnerability.get("advisories", []):
|
||||
url = advisory.get("url") if isinstance(advisory, dict) else None
|
||||
if url:
|
||||
return str(url)
|
||||
return ""
|
||||
|
||||
|
||||
def _build_vulnerability(vulnerability: dict[str, Any], components: dict[str, Component]) -> Vulnerability:
|
||||
"""Flatten a single CycloneDX vulnerability into a ``Vulnerability``.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
components: All components of the document, keyed by bom-ref.
|
||||
|
||||
Returns:
|
||||
The flattened vulnerability.
|
||||
"""
|
||||
return Vulnerability(
|
||||
id=str(vulnerability.get("id") or "Unknown"),
|
||||
severity=_severity_of(vulnerability),
|
||||
description=str(vulnerability.get("description") or vulnerability.get("detail") or ""),
|
||||
cwes=_cwes_of(vulnerability),
|
||||
components=_components_of(vulnerability, components),
|
||||
score=_score_of(vulnerability),
|
||||
recommendation=str(vulnerability.get("recommendation") or ""),
|
||||
url=_url_of(vulnerability),
|
||||
metrics=_metrics_of(vulnerability, _vector_of(vulnerability)),
|
||||
)
|
||||
|
||||
|
||||
def _vector_of(vulnerability: dict[str, Any]) -> str | None:
|
||||
"""Return the newest CVSS vector a CycloneDX vulnerability rates itself with.
|
||||
|
||||
Args:
|
||||
vulnerability: A single CycloneDX vulnerability.
|
||||
|
||||
Returns:
|
||||
The vector, or ``None`` when no rating carries one. CycloneDX names the
|
||||
method rather than versioning the vector, so the ratings are ordered by
|
||||
the method they name and the last generation wins.
|
||||
"""
|
||||
rated = [
|
||||
(str(rating.get("method") or ""), str(rating["vector"]))
|
||||
for rating in vulnerability.get("ratings", [])
|
||||
if isinstance(rating, dict) and rating.get("vector")
|
||||
]
|
||||
|
||||
if not rated:
|
||||
return None
|
||||
|
||||
return max(rated, key=lambda pair: CYCLONEDX_METHODS.get(pair[0].lower(), 0))[1]
|
||||
|
||||
|
||||
def _sort_vulnerabilities(vulnerabilities: list[Vulnerability]) -> list[Vulnerability]:
|
||||
"""Sort vulnerabilities by severity, most severe first."""
|
||||
return sorted(vulnerabilities, key=lambda vuln: -SEVERITY_RANK.get(vuln.severity, 0))
|
||||
|
||||
|
||||
def _to_stats(counts: list[tuple[str, int]], sublabels: dict[str, str] | None = None) -> list[Stat]:
|
||||
"""Turn ``(label, count)`` pairs into ``Stat`` entries with a percentage of the largest count.
|
||||
|
||||
Args:
|
||||
counts: The ``(label, count)`` pairs to turn into stats.
|
||||
sublabels: Optional extra labels keyed by the stat label.
|
||||
|
||||
Returns:
|
||||
The stats, each carrying its share of the largest count in percent.
|
||||
"""
|
||||
if not counts:
|
||||
return []
|
||||
largest = max(count for _, count in counts)
|
||||
return [
|
||||
Stat(
|
||||
label=label,
|
||||
sublabel=(sublabels or {}).get(label, ""),
|
||||
count=count,
|
||||
percent=round(count / largest * 100) if largest else 0,
|
||||
)
|
||||
for label, count in counts
|
||||
]
|
||||
|
||||
|
||||
def _severity_stats(severity_totals: Mapping[str, int]) -> list[Stat]:
|
||||
"""Turn vulnerability counts per severity into stats, in severity order and without empty buckets.
|
||||
|
||||
The percentage is the share of the total number of vulnerabilities, so the bars in the overview
|
||||
show each severity's proportion rather than just its size relative to the largest bucket.
|
||||
|
||||
Args:
|
||||
severity_totals: The number of vulnerabilities per severity.
|
||||
|
||||
Returns:
|
||||
The per-severity stats.
|
||||
"""
|
||||
total = sum(severity_totals.values())
|
||||
return [
|
||||
Stat(
|
||||
label=severity,
|
||||
sublabel="",
|
||||
count=severity_totals[severity],
|
||||
percent=round(severity_totals[severity] / total * 100) if total else 0,
|
||||
)
|
||||
for severity in SEVERITY_ORDER
|
||||
if severity_totals.get(severity)
|
||||
]
|
||||
|
||||
|
||||
def _component_stats(vulnerabilities: list[Vulnerability]) -> list[Stat]:
|
||||
"""Count the components with the most vulnerabilities.
|
||||
|
||||
Args:
|
||||
vulnerabilities: The vulnerabilities to count components from.
|
||||
|
||||
Returns:
|
||||
The most frequently affected components.
|
||||
"""
|
||||
counter: Counter[str] = Counter()
|
||||
for vuln in vulnerabilities:
|
||||
counter.update(vuln.component_names)
|
||||
return _to_stats(counter.most_common(TOP_LIST_SIZE))
|
||||
|
||||
|
||||
def _chart_data(severity_counts: list[Stat], component_counts: list[Stat]) -> dict[str, Any]:
|
||||
"""Build the data blob the Chart.js charts on the scan page are fed with.
|
||||
|
||||
Args:
|
||||
severity_counts: The per-severity stats.
|
||||
component_counts: The per-component stats.
|
||||
|
||||
Returns:
|
||||
The chart data for the severity doughnut and the components bar chart.
|
||||
"""
|
||||
return {
|
||||
"severity": {
|
||||
"labels": [stat.label.capitalize() for stat in severity_counts],
|
||||
"values": [stat.count for stat in severity_counts],
|
||||
"colors": [SEVERITY_COLORS[stat.label] for stat in severity_counts],
|
||||
},
|
||||
"components": {
|
||||
"labels": [stat.label for stat in component_counts],
|
||||
"values": [stat.count for stat in component_counts],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _build_report(vulnerabilities: list[Vulnerability]) -> VdrReport:
|
||||
"""Sort a list of vulnerabilities and derive every statistic the templates render from it.
|
||||
|
||||
Args:
|
||||
vulnerabilities: The vulnerabilities to summarise.
|
||||
|
||||
Returns:
|
||||
The summarised report.
|
||||
"""
|
||||
vulnerabilities = _sort_vulnerabilities(vulnerabilities)
|
||||
severity_counts = _severity_stats(Counter(vuln.severity for vuln in vulnerabilities))
|
||||
component_counts = _component_stats(vulnerabilities)
|
||||
|
||||
return VdrReport(
|
||||
total=len(vulnerabilities),
|
||||
vulnerabilities=vulnerabilities,
|
||||
severity_counts=severity_counts,
|
||||
top_components=component_counts,
|
||||
component_count=len({name for vuln in vulnerabilities for name in vuln.component_names}),
|
||||
chart_data=_chart_data(severity_counts, component_counts),
|
||||
)
|
||||
|
||||
|
||||
# ── Dependency-Track findings export (Finding Packaging Format) ─────────────
|
||||
|
||||
# Dependency-Track spells its severities in upper case and uses ``UNASSIGNED`` for unrated findings.
|
||||
FPF_SEVERITIES: dict[str, str] = {
|
||||
"CRITICAL": "critical",
|
||||
"HIGH": "high",
|
||||
"MEDIUM": "medium",
|
||||
"LOW": "low",
|
||||
"INFO": "info",
|
||||
"UNASSIGNED": "unknown",
|
||||
}
|
||||
|
||||
# The ``references`` field is markdown, so the urls are pulled back out of the link targets.
|
||||
_MARKDOWN_LINK = re.compile(r"\((https?://[^)\s]+)\)")
|
||||
|
||||
|
||||
def _fpf_severity_of(vulnerability: dict[str, Any]) -> str:
|
||||
"""Return the severity of a Dependency-Track finding, ``unknown`` when it carries none."""
|
||||
severity = vulnerability.get("severity")
|
||||
return FPF_SEVERITIES.get(severity.upper(), "unknown") if isinstance(severity, str) else "unknown"
|
||||
|
||||
|
||||
def _fpf_score_of(vulnerability: dict[str, Any]) -> float | None:
|
||||
"""Return the highest CVSS base score a Dependency-Track finding carries.
|
||||
|
||||
Args:
|
||||
vulnerability: The ``vulnerability`` part of a single finding.
|
||||
|
||||
Returns:
|
||||
The highest score of any CVSS version, or ``None`` when the finding carries none.
|
||||
"""
|
||||
scores = [
|
||||
float(vulnerability[key])
|
||||
for key in ("cvssV2BaseScore", "cvssV3BaseScore", "cvssV4Score")
|
||||
if isinstance(vulnerability.get(key), int | float)
|
||||
]
|
||||
return max(scores) if scores else None
|
||||
|
||||
|
||||
def _fpf_cwes_of(vulnerability: dict[str, Any]) -> list[str]:
|
||||
"""Return the CWE ids of a Dependency-Track finding as ``CWE-<n>`` strings.
|
||||
|
||||
Newer reports carry a ``cwes`` list, older ones only a single ``cweId``.
|
||||
|
||||
Args:
|
||||
vulnerability: The ``vulnerability`` part of a single finding.
|
||||
|
||||
Returns:
|
||||
The CWE ids, normalised to the ``CWE-<n>`` form.
|
||||
"""
|
||||
cwes = [
|
||||
f"CWE-{cwe['cweId']}"
|
||||
for cwe in vulnerability.get("cwes", [])
|
||||
if isinstance(cwe, dict) and cwe.get("cweId") is not None
|
||||
]
|
||||
if not cwes and vulnerability.get("cweId") is not None:
|
||||
cwes.append(f"CWE-{vulnerability['cweId']}")
|
||||
return cwes
|
||||
|
||||
|
||||
def _fpf_cwe_names_of(vulnerability: dict[str, Any]) -> dict[str, str]:
|
||||
"""Return what each weakness class of a Dependency-Track finding is called.
|
||||
|
||||
Args:
|
||||
vulnerability: The ``vulnerability`` part of a single finding.
|
||||
|
||||
Returns:
|
||||
The names by identifier. Older reports name only the single ``cweId``
|
||||
they carry, under ``cweName``; newer ones name each entry of ``cwes``.
|
||||
Both are read, and a report that names none simply yields nothing.
|
||||
"""
|
||||
names = {
|
||||
f"CWE-{cwe['cweId']}": str(cwe["name"])
|
||||
for cwe in vulnerability.get("cwes", [])
|
||||
if isinstance(cwe, dict) and cwe.get("cweId") is not None and cwe.get("name")
|
||||
}
|
||||
|
||||
if not names and vulnerability.get("cweId") is not None and vulnerability.get("cweName"):
|
||||
names[f"CWE-{vulnerability['cweId']}"] = str(vulnerability["cweName"])
|
||||
|
||||
return names
|
||||
|
||||
|
||||
def _fpf_url_of(vulnerability: dict[str, Any]) -> str:
|
||||
"""Return the first reference url of a Dependency-Track finding.
|
||||
|
||||
Args:
|
||||
vulnerability: The ``vulnerability`` part of a single finding.
|
||||
|
||||
Returns:
|
||||
A url, or an empty string when the finding references none.
|
||||
"""
|
||||
references = vulnerability.get("references")
|
||||
if not isinstance(references, str):
|
||||
return ""
|
||||
match = _MARKDOWN_LINK.search(references)
|
||||
return match.group(1) if match else ""
|
||||
|
||||
|
||||
def _fpf_component_of(finding: dict[str, Any]) -> Component | None:
|
||||
"""Return the component a Dependency-Track finding was raised against."""
|
||||
component = finding.get("component")
|
||||
if not isinstance(component, dict) or not component.get("name"):
|
||||
return None
|
||||
return Component(
|
||||
name=str(component["name"]),
|
||||
version=str(component.get("version") or ""),
|
||||
latest_version=str(component.get("latestVersion") or ""),
|
||||
)
|
||||
|
||||
|
||||
# The vectors a Dependency-Track export can state, newest generation first.
|
||||
FPF_VECTORS = ("cvssV4Vector", "cvssV3Vector", "cvssV2Vector")
|
||||
|
||||
|
||||
def _fpf_vector_of(vulnerability: dict[str, Any]) -> str | None:
|
||||
"""Return the newest CVSS vector a Dependency-Track finding states, if any."""
|
||||
for key in FPF_VECTORS:
|
||||
if isinstance(vulnerability.get(key), str) and vulnerability[key]:
|
||||
return str(vulnerability[key])
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _metrics_of(vulnerability: dict[str, Any], vector: str | None) -> VulnerabilityMetrics:
|
||||
"""Read one advisory's vector and scoring into the words the panel shows.
|
||||
|
||||
Args:
|
||||
vulnerability: The advisory, in either document's shape.
|
||||
vector: The CVSS vector to read, if the report stated one.
|
||||
|
||||
Returns:
|
||||
The metrics. Everything the report did not state stays empty rather than
|
||||
being filled with a default: a metric nobody stated is not one stated as
|
||||
harmless, and the panel greys out what it was not told.
|
||||
"""
|
||||
reading = parse_vector(vector)
|
||||
spoken = {}
|
||||
|
||||
if reading is not None:
|
||||
confidentiality, integrity, availability = IMPACT_KEYS[reading.generation]
|
||||
spoken = {
|
||||
"attack_vector": _spoken("AV", reading.attack_vector, reading.generation),
|
||||
"attack_complexity": _spoken("AC", reading.attack_complexity, reading.generation),
|
||||
"privileges_required": _spoken(
|
||||
"Au" if reading.generation == 2 else "PR",
|
||||
reading.authentication if reading.generation == 2 else reading.privileges_required,
|
||||
reading.generation,
|
||||
),
|
||||
"user_interaction": _spoken("UI", reading.user_interaction, reading.generation),
|
||||
"confidentiality": _spoken(confidentiality, reading.confidentiality_impact, reading.generation),
|
||||
"integrity": _spoken(integrity, reading.integrity_impact, reading.generation),
|
||||
"availability": _spoken(availability, reading.availability_impact, reading.generation),
|
||||
}
|
||||
|
||||
return VulnerabilityMetrics(
|
||||
epss=_probability(vulnerability.get("epssScore")),
|
||||
epss_percentile=_probability(vulnerability.get("epssPercentile")),
|
||||
published_days_ago=_days_since(vulnerability.get("published")),
|
||||
**spoken,
|
||||
)
|
||||
|
||||
|
||||
def _spoken(metric: str, value: str | None, generation: int) -> str:
|
||||
"""Return the word behind one abbreviation, or nothing where the vector carried none."""
|
||||
return "" if value is None else describe(metric, value, generation)
|
||||
|
||||
|
||||
def _probability(stated: Any) -> float | None: # noqa: ANN401
|
||||
"""Return a probability the report stated, refusing anything that is not one."""
|
||||
return float(stated) if isinstance(stated, int | float) and not isinstance(stated, bool) else None
|
||||
|
||||
|
||||
def _days_since(stated: Any) -> int | None: # noqa: ANN401
|
||||
"""Return how many days ago a timestamp was, or nothing where it cannot be read."""
|
||||
if not isinstance(stated, str):
|
||||
return None
|
||||
|
||||
try:
|
||||
published = datetime.fromisoformat(stated)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
now = datetime.now(published.tzinfo) if published.tzinfo else datetime.now() # noqa: DTZ005
|
||||
|
||||
return max(0, (now - published).days)
|
||||
|
||||
|
||||
def _is_suppressed(finding: dict[str, Any]) -> bool:
|
||||
"""Tell whether a finding was suppressed in Dependency-Track."""
|
||||
analysis = finding.get("analysis")
|
||||
return isinstance(analysis, dict) and bool(analysis.get("isSuppressed"))
|
||||
|
||||
|
||||
def _summarize_findings(findings: list[Any]) -> VdrReport:
|
||||
"""Summarise a Dependency-Track findings export.
|
||||
|
||||
Every finding pairs one component with one vulnerability, so a vulnerability affecting several
|
||||
dependencies arrives as several findings. They are merged back into a single vulnerability that
|
||||
lists all of its components, which is what the scan page shows.
|
||||
|
||||
Args:
|
||||
findings: The ``findings`` list of the export.
|
||||
|
||||
Returns:
|
||||
The summarised report.
|
||||
"""
|
||||
merged: dict[str, Vulnerability] = {}
|
||||
for finding in findings:
|
||||
if not isinstance(finding, dict) or _is_suppressed(finding):
|
||||
continue
|
||||
vulnerability = finding.get("vulnerability")
|
||||
if not isinstance(vulnerability, dict):
|
||||
continue
|
||||
|
||||
key = str(vulnerability.get("uuid") or vulnerability.get("vulnId") or "")
|
||||
component = _fpf_component_of(finding)
|
||||
if key in merged:
|
||||
if component is not None:
|
||||
merged[key].components.append(component)
|
||||
continue
|
||||
|
||||
merged[key] = Vulnerability(
|
||||
id=str(vulnerability.get("vulnId") or "Unknown"),
|
||||
severity=_fpf_severity_of(vulnerability),
|
||||
description=str(vulnerability.get("description") or vulnerability.get("title") or ""),
|
||||
cwes=_fpf_cwes_of(vulnerability),
|
||||
components=[component] if component is not None else [],
|
||||
score=_fpf_score_of(vulnerability),
|
||||
recommendation=str(vulnerability.get("recommendation") or ""),
|
||||
url=_fpf_url_of(vulnerability),
|
||||
metrics=_metrics_of(vulnerability, _fpf_vector_of(vulnerability)),
|
||||
cwe_names=_fpf_cwe_names_of(vulnerability),
|
||||
)
|
||||
|
||||
return _build_report(list(merged.values()))
|
||||
|
||||
|
||||
def summarize_vdr(vdr_data: dict[str, Any]) -> VdrReport:
|
||||
"""Summarise a parsed vulnerability report for rendering on the scan page.
|
||||
|
||||
Both the Dependency-Track findings export and a CycloneDX VDR document are accepted; the
|
||||
``findings`` key is what tells them apart. Every vulnerability of the document is included; the
|
||||
scan page renders the complete list.
|
||||
|
||||
Args:
|
||||
vdr_data: A parsed findings export or CycloneDX VDR document.
|
||||
|
||||
Returns:
|
||||
The summarised report.
|
||||
"""
|
||||
findings = vdr_data.get("findings")
|
||||
if isinstance(findings, list):
|
||||
return _summarize_findings(findings)
|
||||
|
||||
components = _components_by_ref(vdr_data)
|
||||
vulnerabilities = [
|
||||
_build_vulnerability(vuln, components) for vuln in vdr_data.get("vulnerabilities", []) if isinstance(vuln, dict)
|
||||
]
|
||||
return _build_report(vulnerabilities)
|
||||
|
||||
|
||||
@dataclass
|
||||
class VulnerabilitySummary:
|
||||
"""The numbers of one or more vulnerability reports, in a form that can be stored and added up.
|
||||
|
||||
The counterpart of ``FindingsSummary``: the listing pages only show counts, never individual
|
||||
vulnerabilities, so they work off this instead of the full report.
|
||||
"""
|
||||
|
||||
total: int = 0
|
||||
component_count: int = 0
|
||||
severity_totals: dict[str, int] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def severities(self) -> dict[str, int]:
|
||||
"""Vulnerability count per severity, including the severities without any vulnerability."""
|
||||
counts = dict.fromkeys(SEVERITY_ORDER, 0)
|
||||
for severity, count in self.severity_totals.items():
|
||||
if severity in counts:
|
||||
counts[severity] = count
|
||||
return counts
|
||||
|
||||
@property
|
||||
def severity_counts(self) -> list[Stat]:
|
||||
"""Vulnerabilities per severity, in severity order and without empty buckets."""
|
||||
return _severity_stats(self.severities)
|
||||
|
||||
@property
|
||||
def critical_high_count(self) -> int:
|
||||
"""Number of vulnerabilities rated critical or high."""
|
||||
severities = self.severities
|
||||
return severities["critical"] + severities["high"]
|
||||
|
||||
@classmethod
|
||||
def of(cls, report: VdrReport) -> VulnerabilitySummary:
|
||||
"""Reduce a full report to the numbers the listing pages need."""
|
||||
return cls(
|
||||
total=report.total,
|
||||
component_count=report.component_count,
|
||||
severity_totals={stat.label: stat.count for stat in report.severity_counts},
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialise the summary so it can be stored as JSON."""
|
||||
return {
|
||||
"total": self.total,
|
||||
"component_count": self.component_count,
|
||||
"severity_totals": self.severity_totals,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> VulnerabilitySummary:
|
||||
"""Rebuild a summary from stored JSON, tolerating entries that miss a key."""
|
||||
return cls(
|
||||
total=int(data.get("total", 0)),
|
||||
component_count=int(data.get("component_count", 0)),
|
||||
severity_totals=dict(data.get("severity_totals") or {}),
|
||||
)
|
||||
|
||||
|
||||
def merge_vulnerability_summaries(summaries: list[VulnerabilitySummary]) -> VulnerabilitySummary:
|
||||
"""Add up the summaries of several scans.
|
||||
|
||||
Components are counted per scan, so the same dependency in two repositories counts twice - which
|
||||
is what the listing pages want, as those are two different components to update.
|
||||
|
||||
Args:
|
||||
summaries: The summaries to add up.
|
||||
|
||||
Returns:
|
||||
One summary covering all of them.
|
||||
"""
|
||||
severities: Counter[str] = Counter()
|
||||
for summary in summaries:
|
||||
severities.update(summary.severity_totals)
|
||||
|
||||
return VulnerabilitySummary(
|
||||
total=sum(summary.total for summary in summaries),
|
||||
component_count=sum(summary.component_count for summary in summaries),
|
||||
severity_totals=dict(severities),
|
||||
)
|
||||
Reference in new issue
Block a user