324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""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),
|
|
"temporal": _temporal(readings),
|
|
"taxonomy": _taxonomy(readings),
|
|
"affected_surface": _affected_surface(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 _temporal(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:
|
|
age = {"newest": None, "oldest": None, "mean_days": None, "median_days": None, "coverage": 0}
|
|
else:
|
|
dated.sort()
|
|
days = [(datetime.now(UTC) - date.astimezone(UTC)).days for date, _ in dated]
|
|
age = {
|
|
"newest": {"cve": dated[-1][1], "published": dated[-1][0].isoformat(), "days": days[-1]},
|
|
"oldest": {"cve": dated[0][1], "published": dated[0][0].isoformat(), "days": days[0]},
|
|
"mean_days": round(statistics.fmean(days)),
|
|
"median_days": round(statistics.median(days)),
|
|
"coverage": len(dated),
|
|
}
|
|
|
|
return {"published": age, "time_to_exploit_days": None, "cve_volume_trends": None}
|
|
|
|
|
|
def _taxonomy(readings: list[dict[str, Any]]) -> dict[str, Any]:
|
|
cwes: Counter[str] = Counter()
|
|
for one in readings:
|
|
vulnerability = one["finding"]
|
|
values = vulnerability.get("cwes") or []
|
|
if not values and vulnerability.get("cweId") is not None:
|
|
values = [vulnerability["cweId"]]
|
|
for value in values:
|
|
if isinstance(value, dict):
|
|
value = value.get("cweId")
|
|
if value is not None:
|
|
cwes[str(value) if str(value).startswith("CWE-") else f"CWE-{value}"] += 1
|
|
|
|
return {
|
|
"cwe_counts": dict(cwes.most_common()),
|
|
"coverage": sum(cwes.values()),
|
|
"abstraction_level": None,
|
|
"relationships": None,
|
|
}
|
|
|
|
|
|
def _affected_surface(readings: list[dict[str, Any]]) -> dict[str, Any]:
|
|
components = [one["component"] for one in readings]
|
|
names = {str(component.get("name")) for component in components if component.get("name")}
|
|
versions = {str(component.get("version")) for component in components if component.get("version")}
|
|
vendors = set()
|
|
for component in components:
|
|
purl = component.get("purl")
|
|
if isinstance(purl, str) and purl.startswith("pkg:"):
|
|
coordinate = purl[4:].split("@", 1)[0]
|
|
parts = coordinate.split("/")
|
|
if len(parts) > 2 and parts[1]:
|
|
vendors.add(parts[1])
|
|
|
|
return {
|
|
"products": len(names),
|
|
"versions": len(versions),
|
|
"version_ranges": None,
|
|
"vendors": sorted(vendors),
|
|
} |