Allow generation of metrics.json simply by accessing the artifacts folder.
This commit is contained in:
Florian Frank committed
1 parent
7b6c19d8d9
commit
9d7df77923
12 files changed
+802
-486
No files matched your search
@@ -0,0 +1,277 @@
|
||||
"""Build the metrics document from exported vulnerability artifacts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import statistics
|
||||
from collections import Counter
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from .cvss_utils import describe, parse_vector, severity_band, weight
|
||||
from .sources import ReportSources
|
||||
|
||||
SCHEMA_VERSION = "1.0"
|
||||
SCORE_DIGITS = 2
|
||||
PROBABILITY_DIGITS = 5
|
||||
IMPACT_AXES = ("confidentiality", "integrity", "availability")
|
||||
ACCESS_AXES = ("attack_vector", "attack_complexity", "privileges_required", "user_interaction")
|
||||
AXIS_KEYS = {
|
||||
"attack_vector": {2: "AV", 3: "AV", 4: "AV"},
|
||||
"attack_complexity": {2: "AC", 3: "AC", 4: "AC"},
|
||||
"privileges_required": {2: "Au", 3: "PR", 4: "PR"},
|
||||
"user_interaction": {3: "UI", 4: "UI"},
|
||||
"exploit_maturity": {2: "E", 3: "E", 4: "E"},
|
||||
"remediation_level": {2: "RL", 3: "RL"},
|
||||
"confidentiality": {2: "C", 3: "C", 4: "VC"},
|
||||
"integrity": {2: "I", 3: "I", 4: "VI"},
|
||||
"availability": {2: "A", 3: "A", 4: "VA"},
|
||||
}
|
||||
|
||||
|
||||
def build_metrics(sources: ReportSources) -> dict[str, Any]:
|
||||
"""Build the same metrics shape used by persistent storage, from a folder."""
|
||||
readings = _readings(_vdr(sources))
|
||||
document: dict[str, Any] = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"generated_at": datetime.now(UTC).isoformat(),
|
||||
"scan": {"scan_id": str(sources.scan_id), "tool_name": None, "tool_version": None},
|
||||
"sections": [],
|
||||
}
|
||||
if readings:
|
||||
document["sections"].append("dependencies")
|
||||
document["dependencies"] = _dependency_metrics(readings)
|
||||
return document
|
||||
|
||||
|
||||
def _vdr(sources: ReportSources) -> dict[str, Any] | None:
|
||||
content = sources.first("vdr.json", "vdr-2.json", "vdr-3.json")
|
||||
if content is None:
|
||||
return None
|
||||
try:
|
||||
document = json.loads(content)
|
||||
except ValueError:
|
||||
return None
|
||||
return document if isinstance(document, dict) else None
|
||||
|
||||
|
||||
def _readings(document: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||||
if document is None:
|
||||
return []
|
||||
findings = document.get("findings")
|
||||
if isinstance(findings, list):
|
||||
readings = []
|
||||
for finding in findings:
|
||||
if not isinstance(finding, dict) or (finding.get("analysis") or {}).get("isSuppressed"):
|
||||
continue
|
||||
vulnerability = finding.get("vulnerability")
|
||||
component = finding.get("component")
|
||||
if isinstance(vulnerability, dict) and isinstance(component, dict):
|
||||
readings.append(_reading(vulnerability, component, finding))
|
||||
return readings
|
||||
|
||||
components = {
|
||||
item.get("bom-ref"): item
|
||||
for item in document.get("components", [])
|
||||
if isinstance(item, dict) and item.get("bom-ref")
|
||||
}
|
||||
readings = []
|
||||
for vulnerability in document.get("vulnerabilities", []):
|
||||
if not isinstance(vulnerability, dict):
|
||||
continue
|
||||
for affected in vulnerability.get("affects", []):
|
||||
component = components.get(affected.get("ref"), {}) if isinstance(affected, dict) else {}
|
||||
readings.append(_reading(vulnerability, component, {}))
|
||||
return readings
|
||||
|
||||
|
||||
def _reading(vulnerability: dict[str, Any], component: dict[str, Any], occurrence: dict[str, Any]) -> dict[str, Any]:
|
||||
vector_key = next(
|
||||
(key for key in ("cvssV4Vector", "cvssV3Vector", "cvssV2Vector") if vulnerability.get(key)), None
|
||||
)
|
||||
vector = parse_vector(vulnerability.get(vector_key)) if vector_key else None
|
||||
score_key = {
|
||||
"cvssV4Vector": "cvssV4Score",
|
||||
"cvssV3Vector": "cvssV3BaseScore",
|
||||
"cvssV2Vector": "cvssV2BaseScore",
|
||||
}
|
||||
score = vulnerability.get(score_key.get(vector_key, "")) if vector_key else None
|
||||
return {
|
||||
"finding": vulnerability,
|
||||
"component": component,
|
||||
"occurrence": occurrence,
|
||||
"reading": vector,
|
||||
"score": float(score) if isinstance(score, int | float) and not isinstance(score, bool) else None,
|
||||
}
|
||||
|
||||
|
||||
def _dependency_metrics(readings: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
scored = [one for one in readings if one["reading"] is not None]
|
||||
scores = [one["score"] for one in readings]
|
||||
return {
|
||||
"packages": {"affected": len({_package_key(one) for one in readings})},
|
||||
"findings": {"total": len(readings), "scored": len(scored), "suppressed": 0},
|
||||
"most_affected": _most_affected(readings),
|
||||
"scores": {
|
||||
"cvss": _summary(scores) | {"band": severity_band(_mean(scores), 3)},
|
||||
"epss": _summary([one["finding"].get("epssScore") for one in readings], PROBABILITY_DIGITS),
|
||||
"epss_percentile": _summary(
|
||||
[one["finding"].get("epssPercentile") for one in readings], PROBABILITY_DIGITS
|
||||
),
|
||||
},
|
||||
"severity": dict(Counter(_severity(one) for one in readings)),
|
||||
"impact": _impact(scored),
|
||||
"exploitability": _exploitability(scored, readings),
|
||||
"remediation": _remediation(readings),
|
||||
"age": _age(readings),
|
||||
}
|
||||
|
||||
|
||||
def _package_key(one: dict[str, Any]) -> str:
|
||||
component = one["component"]
|
||||
return str(component.get("purl") or component.get("bom-ref") or component.get("name") or "unknown")
|
||||
|
||||
|
||||
def _most_affected(readings: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
groups: dict[str, list[dict[str, Any]]] = {}
|
||||
for one in readings:
|
||||
groups.setdefault(_package_key(one), []).append(one)
|
||||
result = []
|
||||
for key, group in sorted(groups.items(), key=lambda item: len(item[1]), reverse=True)[:10]:
|
||||
component = group[0]["component"]
|
||||
result.append(
|
||||
{
|
||||
"name": component.get("name"),
|
||||
"version": component.get("version"),
|
||||
"latest_version": component.get("latestVersion"),
|
||||
"purl": key,
|
||||
"findings": len(group),
|
||||
"max_cvss": max((one["score"] for one in group if one["score"] is not None), default=None),
|
||||
"max_epss": max(
|
||||
(one["finding"].get("epssScore") for one in group if isinstance(one["finding"].get("epssScore"), int | float)),
|
||||
default=None,
|
||||
),
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _summary(values: list[Any], digits: int = SCORE_DIGITS) -> dict[str, Any]:
|
||||
stated = [float(value) for value in values if isinstance(value, int | float) and not isinstance(value, bool)]
|
||||
if not stated:
|
||||
return {"mean": None, "min": None, "max": None, "p90": None, "coverage": 0}
|
||||
ordered = sorted(stated)
|
||||
position = 0.9 * (len(ordered) - 1)
|
||||
below = int(position)
|
||||
p90 = ordered[below] if below + 1 == len(ordered) else ordered[below] + (ordered[below + 1] - ordered[below]) * (position - below)
|
||||
return {
|
||||
"mean": round(statistics.fmean(stated), digits),
|
||||
"min": round(min(stated), digits),
|
||||
"max": round(max(stated), digits),
|
||||
"p90": round(p90, digits),
|
||||
"coverage": len(stated),
|
||||
}
|
||||
|
||||
|
||||
def _mean(values: list[Any]) -> float | None:
|
||||
stated = [float(value) for value in values if isinstance(value, int | float) and not isinstance(value, bool)]
|
||||
return statistics.fmean(stated) if stated else None
|
||||
|
||||
|
||||
def _severity(one: dict[str, Any]) -> str:
|
||||
reading = one["reading"]
|
||||
band = severity_band(one["score"], reading.generation if reading else 3)
|
||||
return (band or "unscored").lower()
|
||||
|
||||
|
||||
def _weights(readings: list[dict[str, Any]], axis: str) -> list[float]:
|
||||
values = []
|
||||
for one in readings:
|
||||
reading = one["reading"]
|
||||
if reading is None:
|
||||
continue
|
||||
key = AXIS_KEYS[axis].get(reading.generation)
|
||||
value = weight(key, reading.metrics.get(key), reading.generation) if key else None
|
||||
if value is not None:
|
||||
values.append(value)
|
||||
return values
|
||||
|
||||
|
||||
def _axis(readings: list[dict[str, Any]], axis: str) -> dict[str, Any]:
|
||||
counts: Counter[str] = Counter()
|
||||
for one in readings:
|
||||
reading = one["reading"]
|
||||
key = AXIS_KEYS[axis].get(reading.generation) if reading else None
|
||||
if key and reading.metrics.get(key):
|
||||
counts[describe(key, reading.metrics[key], reading.generation)] += 1
|
||||
values = _weights(readings, axis)
|
||||
return {
|
||||
"counts": dict(counts.most_common()),
|
||||
"score": round(statistics.fmean(values) * 10, 2) if values else None,
|
||||
"coverage": len(values),
|
||||
}
|
||||
|
||||
|
||||
def _impact(scored: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
scores = {axis: _axis(scored, axis)["score"] for axis in IMPACT_AXES}
|
||||
stated = {axis: score for axis, score in scores.items() if score is not None}
|
||||
return {
|
||||
"affected": {axis: sum(value > 0 for value in _weights(scored, axis)) for axis in IMPACT_AXES},
|
||||
"scores": scores,
|
||||
"most_affected": max(stated, key=stated.get) if stated else None,
|
||||
"overall": round(statistics.fmean(stated.values()), 2) if stated else None,
|
||||
"coverage": len(_weights(scored, "confidentiality")),
|
||||
}
|
||||
|
||||
|
||||
def _exploitability(scored: list[dict[str, Any]], readings: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
axes = {axis: _axis(scored, axis) for axis in (*ACCESS_AXES, "exploit_maturity")}
|
||||
access = [axes[axis]["score"] for axis in ACCESS_AXES if axes[axis]["score"] is not None]
|
||||
reachable = sum(one["reading"].remotely_reachable for one in scored)
|
||||
return {
|
||||
**axes,
|
||||
"score": round(statistics.fmean(access), 2) if access else None,
|
||||
"remotely_reachable": {
|
||||
"findings": reachable,
|
||||
"share": round(reachable / len(readings), 4) if readings else None,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _remediation(readings: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
upgradeable = [
|
||||
one
|
||||
for one in readings
|
||||
if one["component"].get("latestVersion")
|
||||
and one["component"].get("latestVersion") != one["component"].get("version")
|
||||
]
|
||||
return {
|
||||
"remediation_level": _axis([one for one in readings if one["reading"]], "remediation_level"),
|
||||
"newer_version_available": {
|
||||
"findings": len(upgradeable),
|
||||
"packages": len({_package_key(one) for one in upgradeable}),
|
||||
"share": round(len(upgradeable) / len(readings), 4) if readings else None,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _age(readings: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
dated = []
|
||||
for one in readings:
|
||||
stated = one["finding"].get("published")
|
||||
if isinstance(stated, str):
|
||||
try:
|
||||
dated.append((datetime.fromisoformat(stated.replace("Z", "+00:00")), one["finding"].get("vulnId")))
|
||||
except ValueError:
|
||||
continue
|
||||
if not dated:
|
||||
return {"newest": None, "oldest": None, "mean_days": None, "median_days": None, "coverage": 0}
|
||||
dated.sort()
|
||||
days = [(datetime.now(UTC) - date.astimezone(UTC)).days for date, _ in dated]
|
||||
return {
|
||||
"newest": {"cve": dated[-1][1], "published": dated[-1][0].isoformat(), "days": days[-1]},
|
||||
"oldest": {"cve": dated[0][1], "published": dated[0][0].isoformat(), "days": days[0]},
|
||||
"mean_days": round(statistics.fmean(days)),
|
||||
"median_days": round(statistics.median(days)),
|
||||
"coverage": len(dated),
|
||||
}
|
||||
Reference in new issue
Block a user