From 9d7df779234dd287446a4724c6cc6556e22069bd Mon Sep 17 00:00:00 2001 From: Florian Frank Date: Mon, 21 Sep 2026 21:30:32 +0200 Subject: [PATCH] Allow generation of metrics.json simply by accessing the artifacts folder. --- report_generator/__main__.py | 6 +- report_generator/commands/generate.py | 16 +- report_generator/context.py | 8 +- report_generator/cvss_utils.py | 492 ++++++++++++++++++ report_generator/metrics.py | 277 ++++++++++ report_generator/sources.py | 10 +- report_generator/vendor/README.md | 15 +- report_generator/vendor/__init__.py | 41 +- .../vendor/cc_common/docker/__init__.py | 26 - .../vendor/cc_common/docker/compose_utils.py | 112 ---- .../vendor/cc_common/docker/defs.py | 19 - .../vendor/cc_common/docker/docker_utils.py | 266 ---------- 12 files changed, 802 insertions(+), 486 deletions(-) create mode 100644 report_generator/cvss_utils.py create mode 100644 report_generator/metrics.py delete mode 100644 report_generator/vendor/cc_common/docker/__init__.py delete mode 100644 report_generator/vendor/cc_common/docker/compose_utils.py delete mode 100644 report_generator/vendor/cc_common/docker/defs.py delete mode 100644 report_generator/vendor/cc_common/docker/docker_utils.py diff --git a/report_generator/__main__.py b/report_generator/__main__.py index 0a183e4..13f851c 100644 --- a/report_generator/__main__.py +++ b/report_generator/__main__.py @@ -1,8 +1,8 @@ """Entrypoint for the report generator CLI.""" -from cc_common.api.cli import CLIManager -from cc_common.config import load_environment -from persistent_storage.shared import inject_secrets +from .vendor.cc_common.api.cli import CLIManager +from .vendor.cc_common.config import load_environment +from .vendor.persistent_storage.shared import inject_secrets from .commands.generate import FolderProperties, GenerateProperties, from_folder, generate diff --git a/report_generator/commands/generate.py b/report_generator/commands/generate.py index 21f394f..a50c9a0 100644 --- a/report_generator/commands/generate.py +++ b/report_generator/commands/generate.py @@ -5,17 +5,19 @@ command that demands a database host in order to read a directory is a command n that. What they share is everything after the artifacts are in hand, which is :func:`_report`. """ +import json from pathlib import Path from uuid import UUID from pydantic import BaseModel, Field -from cc_common.api import PipelineError, context -from cc_common.utils import print +from ..vendor.cc_common.api import PipelineError, context +from ..vendor.cc_common.utils import print # Imported for the properties model; the store half of the tool is the only part that needs it. -from persistent_storage.shared import StorageConfiguration +from ..vendor.persistent_storage.shared import StorageConfiguration +from ..metrics import build_metrics from ..rendering import render from ..sources import ReportSources, collect, collect_from_directory @@ -105,10 +107,10 @@ def from_folder(properties: FolderProperties) -> GeneratedReport: PipelineError: The folder cannot be read, names no scan, is missing an input the report needs, or the destination cannot be written. """ - return _report(collect_from_directory(properties.artifacts, properties.scan_id), properties.path) + return _report(collect_from_directory(properties.artifacts, properties.scan_id), properties.path, write_metrics=True) -def _report(sources: ReportSources, path: Path | None) -> GeneratedReport: +def _report(sources: ReportSources, path: Path | None, *, write_metrics: bool = False) -> GeneratedReport: """Render the collected artifacts and write the result. Everything after the artifacts are in hand, which is the same whichever of the two collected @@ -130,6 +132,10 @@ def _report(sources: ReportSources, path: Path | None) -> GeneratedReport: try: destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(content) + if write_metrics: + (destination.parent / "metrics.json").write_text( + json.dumps(build_metrics(sources), indent=2), encoding="utf-8" + ) except OSError as error: message = f"Cannot write the report to '{destination}': {error}" diff --git a/report_generator/context.py b/report_generator/context.py index 1369900..71c5a71 100644 --- a/report_generator/context.py +++ b/report_generator/context.py @@ -20,12 +20,12 @@ from __future__ import annotations from datetime import UTC, datetime from typing import TYPE_CHECKING, Any -from cipherverifier.core.services.sarif import summarize_sarif -from cipherverifier.core.services.vdr import summarize_vdr +from .vendor.cipherverifier.core.services.sarif import summarize_sarif +from .vendor.cipherverifier.core.services.vdr import summarize_vdr if TYPE_CHECKING: - from cipherverifier.core.services.sarif import Finding, SarifReport - from cipherverifier.core.services.vdr import VdrReport, Vulnerability + from .vendor.cipherverifier.core.services.sarif import Finding, SarifReport + from .vendor.cipherverifier.core.services.vdr import VdrReport, Vulnerability from .sources import ReportSources diff --git a/report_generator/cvss_utils.py b/report_generator/cvss_utils.py new file mode 100644 index 0000000..16d798d --- /dev/null +++ b/report_generator/cvss_utils.py @@ -0,0 +1,492 @@ +"""Reads a CVSS vector of any generation and says what it carries. + +Three generations are in circulation at once and a single advisory regularly states two of them, so +nothing here assumes a version -- the vector names its own, and everything after that follows from +the version it named. + +What changed between them, and why one parser cannot pretend they are the same: + +* **v2 (2007)** scores the whole system. It asks how often an attacker has to authenticate (``Au``) + and grades impact as none, partial or complete. +* **v3 (2015, 3.1 in 2019)** scores the vulnerable component instead. ``Au`` becomes ``PR``, which + asks what privileges are needed rather than how many logins; ``UI`` and ``S`` are new; impact + becomes none, low or high; ``AC`` loses its middle value. +* **v4 (2023)** splits impact in two -- the vulnerable system (``VC``/``VI``/``VA``) and whatever + sits behind it (``SC``/``SI``/``SA``) -- which makes ``S`` unnecessary. ``AT`` is new, ``UI`` gains + a passive value, and the temporal group becomes the threat group. + +This module holds no database and no framework, so that the same reading serves persistent_storage +today and archon when the tables move there. Everything that touches a session lives next to the +model instead. + + >>> reading = parse_vector("CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H") + >>> reading.version, reading.reachability, reading.remotely_reachable + ('3.1', 4, True) +""" + +from __future__ import annotations + +from typing import Any, NamedTuple + +# A vector that names no version is a v2 one: v2 is the only generation that did not prefix itself. +DEFAULT_VERSION = "2.0" + +# Which generation a version string belongs to, which is what decides how its metrics are read. +GENERATIONS = {"2.0": 2, "3.0": 3, "3.1": 3, "4.0": 4} + +# The metric that says where an attacker has to be, and the value that puts them anywhere. +FROM_ANYWHERE = "N" + +# What each generation calls "no obstacle at all" for the four metrics an attacker has to get past. +# Answering all four with these is an attack that needs nothing: no position, no effort, no account +# and nobody to fool. v2 has no user interaction metric, so it is scored over three. +# TODO das sollte ich noch refaktoren das ist unübersichtlich +# AV: Access Vector, AC: Access Complexity, Au: Authentication +EASIEST: dict[int, dict[str, str]] = { + 2: {"AV": "N", "AC": "L", "Au": "N"}, + 3: {"AV": "N", "AC": "L", "PR": "N", "UI": "N"}, + 4: {"AV": "N", "AC": "L", "PR": "N", "UI": "N"}, +} + +# What every abbreviation means, per generation, for anything that puts a reading before a person. +# Taken from the specifications; the weights behind them are the scoring formula's business and not +# this module's, which reads a vector rather than recomputing the score from it. +VOCABULARY: dict[int, dict[str, dict[str, str]]] = { + 2: { + "AV": {"L": "Local", "A": "Adjacent network", "N": "Network"}, + "AC": {"H": "High", "M": "Medium", "L": "Low"}, + "Au": {"M": "Multiple", "S": "Single", "N": "None"}, + "C": {"N": "None", "P": "Partial", "C": "Complete"}, + "I": {"N": "None", "P": "Partial", "C": "Complete"}, + "A": {"N": "None", "P": "Partial", "C": "Complete"}, + "E": {"U": "Unproven", "POC": "Proof of concept", "F": "Functional", "H": "High", "ND": "Not defined"}, + "RL": {"OF": "Official fix", "TF": "Temporary fix", "W": "Workaround", "U": "Unavailable", "ND": "Not defined"}, + "RC": {"UC": "Unconfirmed", "UR": "Uncorroborated", "C": "Confirmed", "ND": "Not defined"}, + }, + 3: { + "AV": {"N": "Network", "A": "Adjacent network", "L": "Local", "P": "Physical"}, + "AC": {"L": "Low", "H": "High"}, + "PR": {"N": "None", "L": "Low", "H": "High"}, + "UI": {"N": "None", "R": "Required"}, + "S": {"U": "Unchanged", "C": "Changed"}, + "C": {"N": "None", "L": "Low", "H": "High"}, + "I": {"N": "None", "L": "Low", "H": "High"}, + "A": {"N": "None", "L": "Low", "H": "High"}, + "E": {"X": "Not defined", "U": "Unproven", "P": "Proof of concept", "F": "Functional", "H": "High"}, + "RL": {"X": "Not defined", "O": "Official fix", "T": "Temporary fix", "W": "Workaround", "U": "Unavailable"}, + "RC": {"X": "Not defined", "U": "Unknown", "R": "Reasonable", "C": "Confirmed"}, + }, + 4: { + "AV": {"N": "Network", "A": "Adjacent network", "L": "Local", "P": "Physical"}, + "AC": {"L": "Low", "H": "High"}, + "AT": {"N": "None", "P": "Present"}, + "PR": {"N": "None", "L": "Low", "H": "High"}, + "UI": {"N": "None", "P": "Passive", "A": "Active"}, + "VC": {"H": "High", "L": "Low", "N": "None"}, + "VI": {"H": "High", "L": "Low", "N": "None"}, + "VA": {"H": "High", "L": "Low", "N": "None"}, + "SC": {"H": "High", "L": "Low", "N": "None"}, + "SI": {"H": "High", "L": "Low", "N": "None"}, + "SA": {"H": "High", "L": "Low", "N": "None"}, + "E": {"X": "Not defined", "A": "Attacked", "P": "Proof of concept", "U": "Unreported"}, + }, +} + +# Which key carries the impact on the vulnerable system, per generation. v4 moved it aside to make +# room for a second set covering whatever sits behind that system. +IMPACT_KEYS = {2: ("C", "I", "A"), 3: ("C", "I", "A"), 4: ("VC", "VI", "VA")} + +# The second set, which only v4 has. +SUBSEQUENT_KEYS = ("SC", "SI", "SA") + +# What the base score is called once it is a word. v2 stops at high; v3 added critical, and pushed +# the same vulnerability further up the scale than v2 would have put it. +SEVERITY_BANDS = { + 2: ((0.0, "None"), (4.0, "Low"), (7.0, "Medium"), (10.0, "High")), + 3: ((0.0, "None"), (4.0, "Low"), (7.0, "Medium"), (9.0, "High"), (10.0, "Critical")), + 4: ((0.0, "None"), (4.0, "Low"), (7.0, "Medium"), (9.0, "High"), (10.0, "Critical")), +} + +# What each value is worth when a set of findings has to be averaged into one number per axis. +# +# The base metrics carry the specification's own constants. They are not comparable between +# generations as written -- v2 runs its access vector to 1.0 where v3 stops at 0.85 -- so +# :func:`weight` divides by the largest value of that metric in that generation. What comes back is +# "how far towards the worst case this value sits", 0 to 1, which does compare. +# +# The threat metrics do not carry them. ``E``, ``RL`` and ``RC`` are multipliers in the +# specification, deliberately clustered between 0.91 and 1.0 so that they can only nudge a base +# score. Averaging those would produce a number that is always 0.95 and says nothing, so they are +# ranked by what they actually mean instead: an unproven exploit at 0, one under active attack at 1; +# an official fix at 0, nothing available at 1. Both directions agree with the base metrics -- +# larger is always worse. +WEIGHTS: dict[int, dict[str, dict[str, float]]] = { + 2: { + "AV": {"L": 0.395, "A": 0.646, "N": 1.0}, + "AC": {"H": 0.35, "M": 0.61, "L": 0.71}, + "Au": {"M": 0.45, "S": 0.56, "N": 0.704}, + "C": {"N": 0.0, "P": 0.275, "C": 0.660}, + "I": {"N": 0.0, "P": 0.275, "C": 0.660}, + "A": {"N": 0.0, "P": 0.275, "C": 0.660}, + "E": {"U": 0.0, "POC": 0.5, "F": 0.85, "H": 1.0}, + "RL": {"OF": 0.0, "TF": 0.33, "W": 0.67, "U": 1.0}, + "RC": {"UC": 1.0, "UR": 0.5, "C": 0.0}, + }, + 3: { + "AV": {"P": 0.2, "L": 0.55, "A": 0.62, "N": 0.85}, + "AC": {"H": 0.44, "L": 0.77}, + "PR": {"H": 0.27, "L": 0.62, "N": 0.85}, + "UI": {"R": 0.62, "N": 0.85}, + "C": {"N": 0.0, "L": 0.22, "H": 0.56}, + "I": {"N": 0.0, "L": 0.22, "H": 0.56}, + "A": {"N": 0.0, "L": 0.22, "H": 0.56}, + "E": {"U": 0.0, "P": 0.5, "F": 0.85, "H": 1.0}, + "RL": {"O": 0.0, "T": 0.33, "W": 0.67, "U": 1.0}, + "RC": {"U": 1.0, "R": 0.5, "C": 0.0}, + }, + # v4 scores by lookup table rather than per-metric constants, so there are none to quote. The + # metrics it shares with v3 keep v3's, which preserves their ordering and their spacing; the two + # values v4 added are placed by where the specification ranks them -- ``AT:P`` is an extra + # condition to meet, like a high complexity, and a passive interaction sits between none and an + # active one. + 4: { + "AV": {"P": 0.2, "L": 0.55, "A": 0.62, "N": 0.85}, + "AC": {"H": 0.44, "L": 0.77}, + "AT": {"P": 0.44, "N": 0.85}, + "PR": {"H": 0.27, "L": 0.62, "N": 0.85}, + "UI": {"A": 0.62, "P": 0.73, "N": 0.85}, + "VC": {"N": 0.0, "L": 0.22, "H": 0.56}, + "VI": {"N": 0.0, "L": 0.22, "H": 0.56}, + "VA": {"N": 0.0, "L": 0.22, "H": 0.56}, + "SC": {"N": 0.0, "L": 0.22, "H": 0.56}, + "SI": {"N": 0.0, "L": 0.22, "H": 0.56}, + "SA": {"N": 0.0, "L": 0.22, "H": 0.56}, + "E": {"U": 0.0, "P": 0.5, "A": 1.0}, + }, +} + +# The value every generation uses for "not stated", which is not a value and must not be averaged. +NOT_DEFINED = ("X", "ND") + +# Widths a reading is stored in, so a malformed vector cannot fail an insert. +MAX_VECTOR_LENGTH = 512 +MAX_METRIC_LENGTH = 16 +MAX_VERSION_LENGTH = 8 + + +class CvssReading(NamedTuple): + """What one CVSS vector states, and the two things that follow from it. + + Only the metrics a generation actually has are filled; the rest stay ``None``, because a metric + a version does not define is not the same as one it defines as absent. + + Attributes: + version: The version the vector named, e.g. ``"3.1"``. + generation: ``2``, ``3`` or ``4`` -- which set of metrics applies. + vector: The vector as the report wrote it. + metrics: Every metric of the vector by its abbreviation, including the + environmental and supplemental ones this class has no field for. + attack_vector: ``AV``, in every generation. + attack_complexity: ``AC``, in every generation. + attack_requirements: ``AT``, v4 only. + authentication: ``Au``, v2 only -- how often an attacker has to log in. + privileges_required: ``PR``, v3 and v4 -- what account they need instead. + user_interaction: ``UI``, v3 and v4. + scope: ``S``, v3 only; v4 replaced it with a second impact set. + confidentiality_impact: ``C`` in v2 and v3, ``VC`` in v4. + integrity_impact: ``I`` in v2 and v3, ``VI`` in v4. + availability_impact: ``A`` in v2 and v3, ``VA`` in v4. + subsequent_confidentiality_impact: ``SC``, v4 only. + subsequent_integrity_impact: ``SI``, v4 only. + subsequent_availability_impact: ``SA``, v4 only. + exploit_maturity: ``E`` -- temporal in v2 and v3, threat in v4. + remediation_level: ``RL``, v2 and v3. + report_confidence: ``RC``, v2 and v3. + reachability: How many of the metrics an attacker has to get past sit at + the value that costs them nothing, out of :func:`max_reachability`. + remotely_reachable: Over the network and without an account, which is the + pair that decides whether the internet is in a position to try. + """ + + version: str + generation: int + vector: str + metrics: dict[str, str] + attack_vector: str | None + attack_complexity: str | None + attack_requirements: str | None + authentication: str | None + privileges_required: str | None + user_interaction: str | None + scope: str | None + confidentiality_impact: str | None + integrity_impact: str | None + availability_impact: str | None + subsequent_confidentiality_impact: str | None + subsequent_integrity_impact: str | None + subsequent_availability_impact: str | None + exploit_maturity: str | None + remediation_level: str | None + report_confidence: str | None + reachability: int + remotely_reachable: bool + + +def parse_vector(vector: str | None) -> CvssReading | None: + """Read a CVSS vector of any generation. + + Args: + vector: A vector string, e.g. ``CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H`` + or a v2 one, which carries no prefix: ``AV:N/AC:L/Au:N/C:P/I:P/A:P``. + + Returns: + What the vector states, or ``None`` when there is no vector or it names + none of the metrics its generation defines. ``None`` means "not stated", + which is not the same as "poses no risk", so the two are kept apart. + """ + if not vector: + return None + + version = _version(vector) + generation = GENERATIONS.get(version, 3) + metrics = _metrics(vector, generation) + + if not metrics: + return None + + confidentiality, integrity, availability = IMPACT_KEYS[generation] + + return CvssReading( + version=version, + generation=generation, + vector=vector[:MAX_VECTOR_LENGTH], + metrics=metrics, + attack_vector=metrics.get("AV"), + attack_complexity=metrics.get("AC"), + attack_requirements=metrics.get("AT"), + authentication=metrics.get("Au"), + privileges_required=metrics.get("PR"), + user_interaction=metrics.get("UI"), + scope=metrics.get("S") if generation == 3 else None, + confidentiality_impact=metrics.get(confidentiality), + integrity_impact=metrics.get(integrity), + availability_impact=metrics.get(availability), + subsequent_confidentiality_impact=metrics.get("SC"), + subsequent_integrity_impact=metrics.get("SI"), + subsequent_availability_impact=metrics.get("SA"), + exploit_maturity=metrics.get("E"), + remediation_level=metrics.get("RL"), + report_confidence=metrics.get("RC"), + reachability=sum(metrics.get(key) == value for key, value in EASIEST[generation].items()), + # A metric the vector does not state is not counted as an obstacle either, so an absent one + # leaves the claim unmade rather than making it falsely. + remotely_reachable=_remotely_reachable(metrics, generation), + ) + + +def max_reachability(generation: int) -> int: + """How many obstacles a generation knows about. + + Args: + generation: ``2``, ``3`` or ``4``. + + Returns: + Three for v2, which has no user interaction metric, four otherwise. A + reading is only comparable with another of the same generation, which is + why this is asked for rather than assumed. + """ + return len(EASIEST.get(generation, EASIEST[3])) + + +def severity_band(score: float | None, generation: int) -> str | None: + """Return the word a base score falls under. + + Args: + score: The base score, if the report stated one. + generation: Which scale to read it against -- v2 stops at high, v3 and v4 + added critical, so the same number is not the same word. + + Returns: + The band, or ``None`` when there is no score. + """ + if score is None: + return None + + for limit, name in SEVERITY_BANDS.get(generation, SEVERITY_BANDS[3]): + if score <= limit: + return name + + return SEVERITY_BANDS[generation][-1][1] + + +def weight(metric: str, value: str | None, generation: int = 3) -> float | None: + """Return how far towards the worst case one metric value sits, 0 to 1. + + # TODO hier nochmal das Gewichten explizit prüfen. + This is what makes a set of findings averageable along one axis, and what makes a v2 reading + averageable together with a v4 one: each value is placed between the best and the worst its own + generation allows for that metric, so the result says "how far towards the worst case" rather + than quoting a constant that only means something inside its own formula. + + Stretched across the axis rather than divided by its top, because the specification's constants + are multipliers and do not start at zero: an attack that needs physical access is worth 0.2 of + the 0.85 a network one is worth, and reported as 0.24 it would say a quarter of the worst case + for what is in fact the best case the metric has. Zero here means "as hard as this metric can + make it", one means "as easy". + + Args: + metric: An abbreviation such as ``AV``, ``PR`` or ``VC``. + value: The value the vector carried, if any. + generation: Which generation's scale to read it against. + + Returns: + The position between best and worst case, or ``None`` when the vector + stated nothing, stated "not defined", or stated something the + specification does not list. All three mean the same thing here -- there + is nothing to average -- and are kept apart from a stated zero, which is + a measurement. + """ + if value is None or value in NOT_DEFINED: + return None + + scale = WEIGHTS.get(generation, {}).get(metric, {}) + stated = scale.get(value) + + if stated is None or not scale: + return None + + best, worst = min(scale.values()), max(scale.values()) + + # A metric with a single value says nothing about degree, so it reads as the worst case it has. + return (stated - best) / (worst - best) if worst > best else 1.0 + + +def describe(metric: str, value: str | None, generation: int = 3) -> str: + """Return the word behind one abbreviation. + + Args: + metric: An abbreviation such as ``AV``, ``PR`` or ``VC``. + value: The value the vector carried, if any. + generation: Which vocabulary to read it in. The same letter means + different things across generations -- ``C`` under v2 is "complete", + under v3 it is a metric name -- so this is not optional in practice. + + Returns: + The name the specification gives it, the raw value when the vocabulary + has no entry, or ``"Not stated"`` when the vector did not carry it. + """ + if value is None: + return "Not stated" + + return VOCABULARY.get(generation, {}).get(metric, {}).get(value, value) + + +def _version(vector: str) -> str: + """Return the version a vector names. + + Args: + vector: The vector string. v3 and v4 begin with ``CVSS:``; v2 + begins with its first metric and is recognised by that absence. + + Returns: + The version, defaulting to :data:`DEFAULT_VERSION` for an unprefixed + vector and to the string as written for one this module does not know -- + a version nobody has implemented yet is still worth recording. + """ + head = vector.split("/", 1)[0] + + if not head.upper().startswith("CVSS:"): + return DEFAULT_VERSION + + return head.split(":", 1)[1].strip()[:MAX_VERSION_LENGTH] or DEFAULT_VERSION + + +def _metrics(vector: str, generation: int) -> dict[str, str]: + """Split a vector into its metrics. + + Args: + vector: The vector string. + generation: Which generation it belongs to, which decides whether the + first segment is a version prefix to skip. + + Returns: + The metrics by their abbreviation, spelled as the specification spells + them -- ``Au`` keeps its lower-case second letter, which is the one + abbreviation that is not upper case. A segment that is not a + ``key:value`` pair is ignored rather than raising: a report this store + cannot fully read is still a report worth keeping. + """ + segments = vector.split("/") + metrics: dict[str, str] = {} + + for segment in segments if generation == 2 else segments[1:]: + key, separator, value = segment.partition(":") + + if not separator or not key or not value: + continue + + name = key.strip()[:MAX_METRIC_LENGTH] + metrics["Au" if name.upper() == "AU" and generation == 2 else name.upper()] = value.strip().upper()[ + :MAX_METRIC_LENGTH + ] + + return metrics + + +def _remotely_reachable(metrics: dict[str, str], generation: int) -> bool: + """Whether the vector says an attacker needs neither a position nor an account. + + Args: + metrics: The metrics of the vector. + generation: Which generation states the account requirement, since v2 + asks how often an attacker logs in and v3 onwards what they log in + as. + + Returns: + ``True`` only when the vector states both and both are open. + """ + account = "Au" if generation == 2 else "PR" + + return metrics.get("AV") == FROM_ANYWHERE and metrics.get(account) == "N" + + +def as_row(reading: CvssReading, base_score: float | None = None) -> dict[str, Any]: + """Return a reading shaped like the columns it is stored in. + + Kept here rather than next to the model so that the move to archon is a change of table and not + a change of mapping: whatever holds these columns can be filled from this. + + Args: + reading: What a vector stated. + base_score: The score the report stated alongside the vector, if any. + Not recomputed from the vector -- the report's own number is the one + that has to come back out of the store. + + Returns: + The reading by column name, ready to be handed to a model. + """ + return { + "version": reading.version, + "vector": reading.vector, + "base_score": base_score, + "base_severity": severity_band(base_score, reading.generation), + "attack_vector": reading.attack_vector, + "attack_complexity": reading.attack_complexity, + "attack_requirements": reading.attack_requirements, + "authentication": reading.authentication, + "privileges_required": reading.privileges_required, + "user_interaction": reading.user_interaction, + "scope": reading.scope, + "confidentiality_impact": reading.confidentiality_impact, + "integrity_impact": reading.integrity_impact, + "availability_impact": reading.availability_impact, + "subsequent_confidentiality_impact": reading.subsequent_confidentiality_impact, + "subsequent_integrity_impact": reading.subsequent_integrity_impact, + "subsequent_availability_impact": reading.subsequent_availability_impact, + "exploit_maturity": reading.exploit_maturity, + "remediation_level": reading.remediation_level, + "report_confidence": reading.report_confidence, + "metrics": reading.metrics, + "reachability": reading.reachability, + "remotely_reachable": reading.remotely_reachable, + } diff --git a/report_generator/metrics.py b/report_generator/metrics.py new file mode 100644 index 0000000..1fe7f92 --- /dev/null +++ b/report_generator/metrics.py @@ -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), + } \ No newline at end of file diff --git a/report_generator/sources.py b/report_generator/sources.py index a4b78e2..ec76786 100644 --- a/report_generator/sources.py +++ b/report_generator/sources.py @@ -22,10 +22,10 @@ from uuid import UUID from sqlalchemy import create_engine, text -from cc_common.api import PipelineError, context -from persistent_storage.shared import StorageConfiguration -from persistent_storage.utils.db_utils import get_scan -from persistent_storage.utils.s3_utils import fetch_artifact_files +from .vendor.cc_common.api import PipelineError, context +from .vendor.persistent_storage.shared import StorageConfiguration +from .vendor.persistent_storage.utils.db_utils import get_scan +from .vendor.persistent_storage.utils.s3_utils import fetch_artifact_files if TYPE_CHECKING: from collections.abc import Sequence @@ -33,7 +33,7 @@ if TYPE_CHECKING: from mypy_boto3_s3 import S3Client from sqlalchemy.sql.elements import TextClause - from persistent_storage.wrapper import Artifact + from .vendor.persistent_storage.wrapper import Artifact @dataclass(frozen=True) diff --git a/report_generator/vendor/README.md b/report_generator/vendor/README.md index e92c3ee..fc51a7e 100644 --- a/report_generator/vendor/README.md +++ b/report_generator/vendor/README.md @@ -3,18 +3,9 @@ Copies of everything `report_generator` borrows from the monorepo, so the package can be lifted into a repository of its own. -Nothing in here is edited. `sync.py` writes it, under the same import paths the monorepo uses: - -``` -python -m persistent_storage.report_generator.vendor.sync -``` - -Re-run it after changing anything it lists; the diff then shows what drifted. The copied packages -are removed whole first, so dropping an entry from the list actually drops it from the tree. - -`__init__.py` decides whether these copies are used. It asks whether the real packages can be -imported — not a flag, not an environment variable — so the same code runs in the monorepo and -outside it, and the copies can never shadow the real packages while those are present. +The files in here are committed with the standalone package. `__init__.py` puts +these copies first on `sys.path`, so the package does not accidentally use an +unrelated installed copy of one of these packages. ## What is copied, and what for diff --git a/report_generator/vendor/__init__.py b/report_generator/vendor/__init__.py index e5e3e71..c855590 100644 --- a/report_generator/vendor/__init__.py +++ b/report_generator/vendor/__init__.py @@ -1,53 +1,26 @@ -"""Copies of what this tool borrows, and the switch that makes them the ones it uses. +"""Make the report generator's bundled support packages importable. -In the monorepo the real packages are importable and nothing here is touched. In a repository of -its own they are not, and :func:`activate` puts these copies on the path instead. - -The check is "can it be imported", not a flag or an environment variable, so the same code runs in -both places without being told which one it is in -- and so the copies can never shadow the real -packages while those are present. +The report generator is a standalone repository. Its copies of ``cc_common``, +``persistent_storage`` and ``cipherverifier`` are part of this package and must +win even when similarly named packages happen to be installed on the machine. """ from __future__ import annotations -import importlib.util import sys from pathlib import Path VENDOR_ROOT = Path(__file__).resolve().parent # One module out of each copied package. If these resolve, the real thing is installed. -PROBES = ("cc_common", "persistent_storage.shared", "cipherverifier.core.services.sarif") - - def activate() -> bool: - """Put the vendored copies on the path where the real packages are missing. + """Put the bundled support packages first on the import path. Returns: - Whether the copies were used. ``False`` in the monorepo, where the real - packages are installed and are left to win. + Always ``True``. The return value is retained for callers that used the + previous activation API. """ - if all(_installed(name) for name in PROBES): - return False - if str(VENDOR_ROOT) not in sys.path: sys.path.insert(0, str(VENDOR_ROOT)) return True - - -def _installed(name: str) -> bool: - """Whether a module can be imported without importing it. - - Args: - name: A dotted module name. - - Returns: - Whether the import system can find it. Wrapped because ``find_spec`` - raises rather than returning ``None`` when a parent package is itself - missing, which is exactly the case being tested for. - """ - try: - return importlib.util.find_spec(name) is not None - except (ImportError, ValueError): - return False diff --git a/report_generator/vendor/cc_common/docker/__init__.py b/report_generator/vendor/cc_common/docker/__init__.py deleted file mode 100644 index dea4a24..0000000 --- a/report_generator/vendor/cc_common/docker/__init__.py +++ /dev/null @@ -1,26 +0,0 @@ -"""Collection of CLI helper functions.""" - -from .compose_utils import compose_down, compose_up -from .defs import DockerArchitecture -from .docker_utils import ( - abort_container, - check_image_exist, - create_docker_image, - exec_command, - is_docker_running, - run_command, - run_command_with_input, -) - -__all__ = [ - "DockerArchitecture", - "abort_container", - "check_image_exist", - "compose_down", - "compose_up", - "create_docker_image", - "exec_command", - "is_docker_running", - "run_command", - "run_command_with_input", -] diff --git a/report_generator/vendor/cc_common/docker/compose_utils.py b/report_generator/vendor/cc_common/docker/compose_utils.py deleted file mode 100644 index b1aa9f4..0000000 --- a/report_generator/vendor/cc_common/docker/compose_utils.py +++ /dev/null @@ -1,112 +0,0 @@ -"""Provide utilities for driving Docker Compose stacks. - -Unlike the rest of this package, Compose is not reachable through the docker-py -SDK. It is a separate binary. These helpers are therefore implemented separately, -while keeping the same logging and error conventions as the SDK-based utilities next to them. -""" - -from pathlib import Path -from shutil import which - -from cc_common.api import PipelineError, context -from cc_common.processes import run_subprocess - - -def compose_up(compose_file: Path, project_name: str, *arguments: str) -> None: - """Start the services of a Compose file. - - Nothing is implied about how they start: pass ``"--detach"`` to return once - the containers exist, ``"--build"`` to rebuild images first, and so on. A - call without arguments runs in the foreground, exactly as Compose would. - - Args: - compose_file: Path to the Compose file describing the stack. - project_name: Name of the Compose project to act on. - arguments: Additional options forwarded to ``compose up``. - - Raises: - PipelineError: If Docker is unavailable, the Compose file does not - exist, or Compose exits with a non-zero status. - """ - run_compose(compose_file, project_name, "up", *arguments) - - -def compose_down(compose_file: Path, project_name: str, *arguments: str) -> None: - """Stop the services of a Compose file and remove their containers. - - Named volumes survive unless ``"--volumes"`` is passed, which is the one - argument here that destroys data — Compose's own default is to keep it, and - this function does not second-guess that either way. - - Args: - compose_file: Path to the Compose file describing the stack. - project_name: Name of the Compose project to act on. - arguments: Additional options forwarded to ``compose down``. - - Raises: - PipelineError: If Docker is unavailable, the Compose file does not - exist, or Compose exits with a non-zero status. - """ - run_compose(compose_file, project_name, "down", *arguments) - - -def run_compose(compose_file: Path, project_name: str, *arguments: str) -> None: - """Run any Docker Compose subcommand against a Compose file. - - Args: - compose_file: Path to the Compose file describing the stack. - project_name: Name of the Compose project to act on. Passing it - explicitly keeps containers and volumes stable no matter which - directory the caller is invoked from — Compose would otherwise - derive the name from the directory holding the Compose file. - arguments: The Compose subcommand and its options, e.g. ``"up", - "--detach"``. - - Raises: - PipelineError: If Docker is unavailable, the Compose file does not - exist, or Compose exits with a non-zero status. - """ - if not compose_file.is_file(): - raise PipelineError(f"Compose file not found at {compose_file}") - - command = compose_command() - subcommand = " ".join(arguments) - - context.logger.info(f"Running 'compose {subcommand}' on {compose_file} ...") - - # Currently, no solution without using subprocesses available. - exit_code = run_subprocess( - [*command, "--project-name", project_name, "--file", str(compose_file), *arguments], - Path.cwd(), - stream_output=True, - on_stdout=context.logger.info, - on_stderr=context.logger.info, - fill_stream_threaded=True, - ) - - if exit_code != 0: - raise PipelineError(f"'compose {subcommand}' failed with exit code {exit_code}") - - -def compose_command() -> list[str]: - """Determine how Docker Compose can be invoked on this machine. - - Returns: - The executable and subcommand to prefix the Compose arguments with, - preferring the ``docker compose`` plugin over the standalone binary. - - Raises: - PipelineError: If neither ``docker`` nor the legacy ``docker-compose`` - executable is on the PATH. - """ - docker_path = which("docker") - - if docker_path is not None: - return [docker_path, "compose"] - - legacy_compose_path = which("docker-compose") - - if legacy_compose_path is None: - raise PipelineError("Docker executable not found") - - return [legacy_compose_path] diff --git a/report_generator/vendor/cc_common/docker/defs.py b/report_generator/vendor/cc_common/docker/defs.py deleted file mode 100644 index 4f094a9..0000000 --- a/report_generator/vendor/cc_common/docker/defs.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Docker-specific architecture and platform definitions.""" - -from enum import Enum - -from cc_common.defines import CPUArchitecture, OperatingSystem - -CONTAINER_DEFAULT_ABORT_TIMEOUT_IN_S = 5 - - -class DockerArchitecture(Enum): - """Enumeration of supported Docker platforms combining OS and CPU architecture.""" - - LINUX_AMD64 = (OperatingSystem.LINUX, CPUArchitecture.AMD64) - LINUX_ARM64 = (OperatingSystem.LINUX, CPUArchitecture.ARM64) - - @property - def arch_str(self) -> str: - """Return the os/cpu string which can be passed to docker command as plattform.""" - return f"{self.value[0].value}/{self.value[1].value}" diff --git a/report_generator/vendor/cc_common/docker/docker_utils.py b/report_generator/vendor/cc_common/docker/docker_utils.py deleted file mode 100644 index 79ccb1e..0000000 --- a/report_generator/vendor/cc_common/docker/docker_utils.py +++ /dev/null @@ -1,266 +0,0 @@ -"""Provide utility functions for interacting with Docker via the Docker SDK. - -Supports image management, container execution, and environment validation -through the official docker-py client with logging integration. -""" - -import os -import shlex -import socket -from pathlib import Path -from typing import Any - -import docker -import docker.errors -from docker import DockerClient - -from cc_common.api import context - -from .defs import CONTAINER_DEFAULT_ABORT_TIMEOUT_IN_S, DockerArchitecture - - -def _get_client() -> DockerClient: - """Create and return a Docker client from the local environment. - - Raises: - docker.errors.DockerException: If the Docker daemon is unreachable. - """ - return docker.from_env() - - -def _stop_container(container_id: str, timeout_in_s: int) -> None: - """Stops a specific container by ID.""" - client = _get_client() - try: - container = client.containers.get(container_id) - context.logger.info(f"Aborting and removing container: {container_id}") - container.stop(timeout=timeout_in_s) - container.remove(force=True) - except docker.errors.NotFound: - context.logger.debug(f"Container {container_id} already removed.") - except Exception as e: - context.logger.error(f"Failed to abort container {container_id}: {e}") - - -def is_docker_running() -> bool: - """Check whether the Docker daemon is running.""" - context.logger.info("Checking if Docker is running") - try: - _get_client().ping() - return True - except docker.errors.DockerException: - return False - - -def check_image_exist(image_name: str) -> bool: - """Check whether a Docker image exists locally.""" - context.logger.info(f"Checking if image '{image_name}' exists") - try: - _get_client().images.get(image_name) - return True - except docker.errors.ImageNotFound: - return False - - -def create_docker_image( - image_name: str, - project_folder: Path, - docker_architecture: DockerArchitecture | None = None, - build_dargs: dict[str, str] | None = None, -) -> None: - """Build a Docker image from the specified project directory.""" - context.logger.info(f"Creating docker image '{image_name}' from '{project_folder}'") - - client = _get_client() - - try: - for chunk in client.api.build( - path=str(project_folder), - tag=image_name, - buildargs=build_dargs, - decode=True, # Low-level API handles this correctly - rm=True, - platform=docker_architecture.arch_str if docker_architecture else None, - ): - if not isinstance(chunk, dict): - continue - - if "stream" in chunk: - message = str(chunk["stream"]).strip() - if message: - context.logger.debug(message) - - if "error" in chunk: - error_data = chunk.get("errorDetail", {}) - error_msg = error_data.get("message", chunk["error"]) - - raise docker.errors.BuildError(error_msg, iter([chunk])) - - except docker.errors.BuildError as e: - context.logger.error(f"Docker build failed: {e}") - raise e - except Exception as e: - context.logger.error(f"An unexpected error occurred during build: {e}") - raise e - - -def _rewrite_path_in_cmd(cmd: list[str], old_path: str, new_path: str) -> list[str]: - """Replace `old_path` with `new_path` in every argument that points at it.""" - rewritten = [] - for arg in cmd: - if arg == old_path: - rewritten.append(new_path) - elif arg.startswith(f"{old_path}/"): - rewritten.append(f"{new_path}{arg[len(old_path) :]}") - else: - rewritten.append(arg) - return rewritten - - -def _wrap_cmd_with_copy(cmd: list[str], mount_folders: list[tuple[Path, str]], target_folder: str) -> list[str]: - """Wrap `cmd` so the mounted folders are copied into `target_folder` first. - - The copy is created by the container's own user, so it is writable without any - privilege escalation, and the read-only mounts stay untouched. Arguments in `cmd` - that point into a mounted folder are rewritten to the copy. Requires `sh`, `mkdir` - and `cp` in the image. - """ - copy_cmds = [] - for _, container_path in mount_folders: - target_path = f"{target_folder.rstrip('/')}/{Path(container_path).name}" - quoted_target = shlex.quote(target_path) - copy_cmds.append(f"mkdir -p {quoted_target} && cp -a {shlex.quote(container_path)}/. {quoted_target}/") - cmd = _rewrite_path_in_cmd(cmd, container_path, target_path) - - script = " && ".join([*copy_cmds, f"exec {shlex.join(cmd)}"]) - return ["sh", "-c", script] - - -def run_command( - image_name: str, - cmd: list[str], - mount_folders: list[tuple[Path, str]], - remove_after_execution: bool, - copy_mounts_to: str | None = None, -) -> str: - """Run a command inside a Docker container with optional volume mounts. - - Args: - image_name: Image to run the command in. - cmd: Command and arguments to execute. - mount_folders: Pairs of (host path, container path) to mount. - remove_after_execution: Whether the container is removed after it exits. - copy_mounts_to: Optional container folder (for example the user's home) the - mounted folders are copied into before the command runs. The mounts are - then read-only and `cmd` arguments pointing at them are rewritten to the - copy, so the command works on data owned by the container's own user. - """ - mount_mode = "ro" if copy_mounts_to else "rw" - volumes = { - str(local_path.resolve()): {"bind": container_path, "mode": mount_mode} - for local_path, container_path in mount_folders - } - - context.logger.info(f"Running command {cmd} in image '{image_name}'") - - run_cmd = _wrap_cmd_with_copy(cmd, mount_folders, copy_mounts_to) if copy_mounts_to else cmd - - output = _get_client().containers.run( - image=image_name, command=run_cmd, volumes=volumes, remove=remove_after_execution, stdout=True, stderr=True - ) - return output.decode("utf-8") - - -def run_command_with_input( - image_name: str, cmd: list[str], input_data: str, environment: dict[str, Any] | None = None -) -> str: - """Run a command inside a Docker container, streaming a string directly to its stdin.""" - context.logger.info(f"Running command {cmd} in image '{image_name}'") - client = _get_client() - - container = client.containers.create( - image=image_name, command=cmd, stdin_open=True, tty=False, environment=environment - ) - - try: - container.start() - - docker_sock = container.attach_socket(params={"stdin": 1, "stream": 1}) - - try: - fd = docker_sock.fileno() # type: ignore[union-attr] - clean_fd = os.dup(fd) - - if hasattr(docker_sock, "_response"): - try: # noqa: SIM105 - docker_sock._response.close() - except Exception as e: - context.logger.error(f"hasattr returns {e}") - docker_sock._response = None # drop the reference entirely - - family = socket.AF_UNIX if fd < 100 else socket.AF_INET - clean_sock = socket.socket(family=family, fileno=clean_fd) - - finally: - docker_sock.close() - try: - clean_sock.sendall(input_data.encode("utf-8")) - clean_sock.shutdown(socket.SHUT_WR) - finally: - clean_sock.close() - - exit_result = container.wait() - output = container.logs(stdout=True, stderr=True).decode("utf-8") - - if exit_result["StatusCode"] != 0: - raise RuntimeError(f"Container failed ({exit_result['StatusCode']}):\n{output}") - - return output - finally: - container.remove(force=True) - - -def exec_command(container_id: str, cmd: list[str]) -> str: - """Execute a command inside a running container. - - Returns: - The stdout output of the command as a decoded string. - """ - context.logger.info(f"Executing command {cmd} in container '{container_id}'") - - container = _get_client().containers.get(container_id) - exit_code, output = container.exec_run(cmd) - - if exit_code != 0: - raise docker.errors.APIError(f"Command {cmd} failed in container '{container_id}' with exit code {exit_code}") - - if isinstance(output, bytes): - return output.decode("utf-8") - - return b"".join(output).decode("utf-8") - - -def abort_container( - container_id: str | None = None, - image_name: str | None = None, - timeout_in_s: int = CONTAINER_DEFAULT_ABORT_TIMEOUT_IN_S, -) -> None: - """Forcefully stop and remove a specific container by ID.""" - if not container_id and not image_name: - context.logger.error("Neither container nor image specified.") - return - - client = _get_client() - if container_id: - _stop_container(container_id, timeout_in_s) - else: - running_containers = client.containers.list() - for container in running_containers: - if ( - image_name - and container.id - and container.image - and container.image.tags - and any(image_name in tag for tag in container.image.tags) - ): - _stop_container(container.id, timeout_in_s)