"""Turns a scan's stored artifacts into the context the report template draws. No parsing of its own. The SARIF and the vulnerability report are read by the summarisers in :mod:`cipherverifier.core.services.sarif` and :mod:`cipherverifier.core.services.vdr` -- both of which are deliberately free of Django and of any store -- and the pipeline metadata by the helpers in :mod:`persistent_storage.utils.rep_utils`. What is left here is the mapping: which summarised number goes in which row, and what to say where the artifacts do not say anything. That last part is the rule this module is built around. A field the artifacts cannot answer is written as :data:`UNKNOWN` rather than guessed, left blank or quietly dropped. A reader who sees ```` knows the pipeline did not report it; a reader who sees a plausible zero does not. Moving this to archon is a change of what :mod:`sources` reads from. The two summarisers travel with it -- they belong beside whatever holds the artifacts, not inside the web application that happens to import them today. """ from __future__ import annotations import re from collections import Counter from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from .vendor.cipherverifier.core.services.sarif import summarize_sarif from .vendor.cipherverifier.core.services.vdr import summarize_vdr from .layout import TITLE if TYPE_CHECKING: from .vendor.cipherverifier.core.services.sarif import Finding, SarifReport from .vendor.cipherverifier.core.services.vdr import VdrReport, Vulnerability from .sources import ReportSources # What stands in for a field the stored artifacts do not answer. UNKNOWN = "" # How many entries the summary lists carry. The sections themselves list every finding. MOST_CRITICAL = 6 BARS = 10 # The severity a SARIF level is reported under. SARIF names four levels and the report bands five, # so the mapping is written down rather than inferred from the spelling. SEVERITY_OF_LEVEL = { "error": "Critical", "warning": "High", "note": "Medium", "none": "Low", "info": "Low", } SEVERITY_ORDER = ("Critical", "High", "Medium", "Low") def build_context(sources: ReportSources, documents: dict[str, Any]) -> dict[str, Any]: """Build the report context from a scan's parsed artifacts. Args: sources: The scan and its stored files. documents: The parsed artifacts by role -- ``metadata``, ``findings``, ``vulnerabilities`` -- each absent where the scan stored none. Returns: The context :func:`layout.render_pdf` draws. Sections whose artifact is missing are left out entirely, which the template renders as a shorter report. """ metadata = documents.get("metadata") or {} tools = _tools(documents.get("findings")) sca = summarize_vdr(documents["vulnerabilities"]) if documents.get("vulnerabilities") else None context: dict[str, Any] = { "cover": _cover(sources, metadata, tools, sca), "summary": _summary(tools, sca, documents.get("metrics")), "sections": [_tool_section(name, report) for name, report in tools], } # Without the pipeline's metadata these two pages would be nothing but gaps. if metadata: context["scan"] = _scan(sources, metadata) context["project"] = _project(metadata) critical = _critical(tools, sca) if critical["table"]["rows"]: context["critical"] = critical if sca is not None: context["sections"].append(_sca_section(sca)) set_aside = _set_aside_section(tools) if set_aside is not None: context["sections"].append(set_aside) context["contents"] = _contents(context) return context def _tools(sarif: Any) -> list[tuple[str, SarifReport]]: # noqa: ANN401 """The SARIF document split by the tool that produced each run, in the order the runs appear.""" if not isinstance(sarif, dict): return [] runs = [run for run in sarif.get("runs") or [] if isinstance(run, dict)] return [(_tool_name(run), summarize_sarif({"runs": [run]})) for run in runs] def _tool_name(run: dict[str, Any]) -> str: """What the tool that produced a run calls itself.""" driver = (run.get("tool") or {}).get("driver") or {} return str(driver.get("name") or driver.get("fullName") or UNKNOWN) def _contents(context: dict[str, Any]) -> list[dict[str, str]]: """The table of contents: every top-level part of the report that is actually in it.""" parts = [context[key] for key in ("scan", "project", "summary", "critical") if key in context] return [{"id": part["id"], "title": part["title"]} for part in [*parts, *context["sections"]]] # --- the page ------------------------------------------------------------------------------ def _fact(label: str, value: Any, *, gap: bool = False, kind: str | None = None) -> dict[str, Any]: # noqa: ANN401 """One label/value row, with anything unstated written as :data:`UNKNOWN`.""" return {"label": label, "value": _stated(value), "gap": gap, "kind": kind} def _row(label: str, value: Any, **marks: Any) -> dict[str, Any]: # noqa: ANN401 """One row of a card.""" return {"label": label, "value": _stated(value), **marks} def _stated(value: Any) -> str: # noqa: ANN401 """What a value reads as, or :data:`UNKNOWN` where there is nothing to read. Zero is a reading and survives; an empty string and ``None`` are not, and do not. The difference matters on a page where both appear: "0 filtered" is a measurement, and a blank is a gap. """ if value is None or (isinstance(value, str) and not value.strip()): return UNKNOWN return str(value) def _cover( sources: ReportSources, metadata: dict[str, Any], tools: list[tuple[str, SarifReport]], sca: VdrReport | None, ) -> dict[str, Any]: """The first page.""" configuration = metadata.get("configuration") or {} technologies = [name for name, _ in tools] + (["SCA"] if sca is not None else []) return { "scan_id": str(sources.scan_id), "subject": _stated(configuration.get("repository_name")), "footer_date": datetime.now(UTC).strftime("%d %B %Y"), "logo": "cc_logo.png", "title": TITLE, "identity": [ _fact("Scan Identifier", sources.scan_id, kind="mono"), _fact("Technology", ", ".join(technologies) or None), _fact("Project", configuration.get("repository_name"), gap=True), _fact("Version", configuration.get("repository_version")), _fact("Report Created", datetime.now(UTC).strftime("%A, %B %d, %Y")), _fact("Manufacturer", None), ], "severity_ledger": _severity_ledger(tools, sca), } def _severity_ledger(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> dict[str, Any] | None: """How many issues of each severity survived filtering, over every tool and the dependencies.""" counts = dict.fromkeys(SEVERITY_ORDER, 0) for _, report in tools: for finding in _kept(report): counts[_finding_severity(finding)] += 1 for vulnerability in sca.vulnerabilities if sca else []: counts[_vdr_severity(vulnerability)] += 1 if not any(counts.values()): return None rows: list[dict[str, Any]] = [ {"label": f"{name} severity issues", "count": counts.get(name, 0)} for name in SEVERITY_ORDER ] rows.append({"label": "Total true issues", "count": sum(counts.values()), "total": True}) return {"title": "True issues by severity", "rows": rows} def _scan(sources: ReportSources, metadata: dict[str, Any]) -> dict[str, Any]: """What the pipeline did, stage by stage.""" stages = [stage for stage in metadata.get("stages") or [] if isinstance(stage, dict)] failed = [stage["label"] for stage in stages if stage.get("status") not in ("success", None)] return { "id": "scan", "title": "CipherCell Scan Information", "overview": [ _fact("Scan started at", _moment(metadata.get("pipeline_start_iso"))), _fact("Scan duration", _duration(metadata.get("pipeline_duration_seconds"))), _fact("Scan completed at", _moment(metadata.get("pipeline_end_iso"))), _fact( "Pipeline Steps", ", ".join(_stated(stage.get("label") or stage.get("key")) for stage in stages) or None ), _fact("Errors Detected", ", ".join(failed) if failed else "None" if stages else None), ], "stages": [_stage(stage, sources) for stage in stages], } def _stage(stage: dict[str, Any], sources: ReportSources) -> dict[str, Any]: """One stage of the pipeline. The tools a stage ran and the profile it ran them under are not recorded in the metadata, so they are left unstated. What is recorded is when it ran, how it ended, and -- indirectly -- what it produced, which is read off the names of the files the scan stored. """ return { "title": stage.get("label") or stage.get("key") or UNKNOWN, "rows": [ _fact("Selected Tools", None), _fact("Profile", None), _fact("Duration", _duration(stage.get("duration_seconds"))), _fact("Created Artifacts", ", ".join(sorted(sources.files)) or None, kind="mono"), _fact("Errors Detected", "None" if stage.get("status") == "success" else stage.get("status")), ], } def _project(metadata: dict[str, Any]) -> dict[str, Any]: """What was scanned.""" configuration = metadata.get("configuration") or {} return { "id": "project", "title": "Project Information", "overview": [ _fact("Project", configuration.get("repository_name")), _fact("Version", configuration.get("repository_version")), # Nothing in the pipeline counts lines, so nothing here can report them. _fact("Analysed lines of code", None), ], # Left as one unstated row rather than a table of zeroes: no stage reports a language # breakdown, and a share invented from the file extensions of the findings would be a # measurement of where the bugs are, not of what the project is written in. "languages": {"title": "Languages", "rows": [{"label": UNKNOWN, "share": UNKNOWN}]}, } def _summary( tools: list[tuple[str, SarifReport]], sca: VdrReport | None, metrics: dict[str, Any] | None ) -> dict[str, Any]: """The counts, and where they concentrate.""" sca_total = sca.total if sca else 0 raw = sum(report.total for _, report in tools) + sca_total true_issues = [f"{len(_kept(report))} {name}" for name, report in tools] + ([f"{sca_total} SCA"] if sca else []) summary = { "id": "summary", "title": "Summary", "overview": [ _fact("Raw findings", raw or None), # Only SARIF findings can be set aside. A vulnerability suppressed in Dependency-Track # never reaches the report at all, so the two cannot be added up. _fact( "False positives / suppressed", ", ".join(f"{len(_set_aside(report))} {name}" for name, report in tools) or None, ), _fact("True issues", ", ".join(true_issues) or None), _fact("Distinct issue types (true)", ", ".join(_issue_types(tools, sca)) or None), ], "distributions": _distributions(tools, sca), } dependencies = metrics.get("dependencies") if isinstance(metrics, dict) else None if isinstance(dependencies, dict): summary["metrics"] = _metric_cards(dependencies) return summary def _metric_cards(dependencies: dict[str, Any]) -> list[dict[str, Any]]: """Turn the metrics document into compact summary cards.""" scores = dependencies.get("scores") or {} impact = dependencies.get("impact") or {} exploitability = dependencies.get("exploitability") or {} temporal = dependencies.get("temporal") or {} published = temporal.get("published") or {} remediation = dependencies.get("remediation") or {} upgrades = remediation.get("newer_version_available") or {} surface = dependencies.get("affected_surface") or {} taxonomy = dependencies.get("taxonomy") or {} return [ { "title": "Severity & impact", "note": "CVSS is the mean base score of the scored dependency findings. Impact values are normalized averages from 0 to 10; higher means greater potential damage.", "rows": [ _fact("Average CVSS score (0-10)", _metric(scores.get("cvss"), "mean")), _fact("Average CVSS severity", (scores.get("cvss") or {}).get("band"), kind="severity"), _fact("Average confidentiality impact (0-10)", _metric(impact.get("scores"), "confidentiality")), _fact("Average integrity impact (0-10)", _metric(impact.get("scores"), "integrity")), _fact("Average availability impact (0-10)", _metric(impact.get("scores"), "availability")), ], }, { "title": "Exploitability", "note": "These are normalized averages of the CVSS vector dimensions across scored findings. Higher values indicate fewer obstacles for an attacker, not a probability of exploitation.", "rows": [ _fact( label, _metric(exploitability.get(axis), "score"), ) for axis, label in ( ("attack_vector", "Average attack vector exposure (0-10)"), ("attack_complexity", "Average attack complexity exposure (0-10)"), ("privileges_required", "Average privilege exposure (0-10)"), ("user_interaction", "Average user-interaction exposure (0-10)"), ) ], }, { "title": "Temporal", "note": "Published dates and vulnerability age are calculated from the advisory timestamps. Time-to-exploit and volume trends require historical exploitation or scan data that this export does not contain.", "rows": [ _fact("Newest published", _published(published.get("newest"))), _fact("Mean vulnerability age", _days(published.get("mean_days"))), _fact("Median vulnerability age", _days(published.get("median_days"))), _fact("Time to exploit", temporal.get("time_to_exploit_days")), _fact("CVE volume trends", temporal.get("cve_volume_trends")), ], }, { "title": "Surface & taxonomy", "note": "Surface counts describe the packages represented by the findings. CWE coverage counts findings with a CWE classification; it does not measure CWE quality or abstraction depth.", "rows": [ _fact("Affected dependencies", surface.get("products")), _fact("Potential remedies", _remedy(upgrades)), _fact("CWE coverage", taxonomy.get("coverage")), _fact("CWE abstraction level", taxonomy.get("abstraction_level")), _fact("CWE relationships", taxonomy.get("relationships")), ], }, ] def _metric(value: Any, key: str) -> Any: # noqa: ANN401 return value.get(key) if isinstance(value, dict) else None def _published(value: Any) -> Any: # noqa: ANN401 return value.get("cve") if isinstance(value, dict) else None def _days(value: Any) -> str | None: # noqa: ANN401 return f"{value} days" if isinstance(value, int | float) else None def _remedy(value: dict[str, Any]) -> str | None: findings = value.get("findings") packages = value.get("packages") if isinstance(findings, int) and isinstance(packages, int): return f"{findings} findings across {packages} packages upgradeable" return None def _issue_types(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> list[str]: """How many distinct kinds of true issue each tool, and the dependency check, found.""" counted = [f"{len({one.rule_id for one in _kept(report)})} {name} rules" for name, report in tools] if sca is not None: counted.append(f"{sca.component_count} affected components") return counted def _distributions(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> list[dict[str, Any]]: """Where the issues concentrate: by rule for each SARIF tool, by package for SCA.""" blocks = [] for name, report in tools: findings = _kept(report) if findings: blocks.append( { "title": f"True {name} issues by rule ({len(findings)})", **_bars(_group(findings, lambda one: one.rule_id or UNKNOWN, _finding_severity)), } ) if sca and sca.vulnerabilities: blocks.append( { "title": f"Most affected components, true SCA issues ({sca.component_count})", **_bars( _group( [(one, name) for one in sca.vulnerabilities for name in one.component_names], lambda pair: pair[1], lambda pair: _vdr_severity(pair[0]), ), mono=True, ), } ) return blocks def _group[T]( items: list[T], label_of: Any, # noqa: ANN401 severity_of: Any, # noqa: ANN401 ) -> list[tuple[str, dict[str, int]]]: """Count items by label, and within each label by severity.""" grouped: dict[str, dict[str, int]] = {} for item in items: counts = grouped.setdefault(label_of(item), dict.fromkeys(SEVERITY_ORDER, 0)) counts[severity_of(item)] += 1 return sorted(grouped.items(), key=lambda pair: -sum(pair[1].values())) def _bars(grouped: list[tuple[str, dict[str, int]]], *, mono: bool = False) -> dict[str, Any]: """Turn grouped counts into bars scaled against the busiest row. Only the busiest :data:`BARS` are drawn, and the caveat says so where the rest were left off. Without that the legend counts what is on the page while the heading counts the report, and the two quietly disagree by however much the tail held. """ shown = grouped[:BARS] largest = max((sum(counts.values()) for _, counts in shown), default=0) or 1 rows: list[dict[str, Any]] = [] totals = dict.fromkeys(SEVERITY_ORDER, 0) for label, counts in shown: rows.append( { "label": label, "count": sum(counts.values()), "mono": mono, "parts": [ {"severity": name, "share": round(counts[name] / largest * 100, 1)} for name in SEVERITY_ORDER if counts[name] ], } ) for name, count in counts.items(): totals[name] += count return { "rows": rows, "legend": { "entries": [ {"severity": name, "label": f"{name} ({totals[name]})"} for name in reversed(SEVERITY_ORDER) if totals[name] ], "caveat": _caveat(shown, grouped, largest), }, } def _caveat(shown: list[Any], grouped: list[Any], largest: int) -> str: """What the bars leave out, and what their length is measured against.""" scale = f"bar length is proportional to the issue count (100 % = {largest})." if len(shown) == len(grouped): return scale return f"{len(shown)} of {len(grouped)} shown; {scale}" def _critical(tools: list[tuple[str, SarifReport]], sca: VdrReport | None) -> dict[str, Any]: """The handful worth reading first, whichever tool found them.""" rows: list[tuple[str, dict[str, Any]]] = [] for name, report in tools: for finding in _kept(report): severity = _finding_severity(finding) rows.append( ( severity, { "cells": [ name, {"value": severity, "kind": "severity"}, _stated(finding.rule_id), {"value": _stated(_where(finding)), "kind": "mono"}, _stated(_text(finding.message)), ] }, ) ) for vulnerability in sca.vulnerabilities if sca else []: severity = _vdr_severity(vulnerability) rows.append( ( severity, { "cells": [ "SCA", {"value": severity, "kind": "severity"}, _stated(vulnerability.id), {"value": _stated(", ".join(vulnerability.component_names)), "kind": "mono"}, _stated(vulnerability.description), ] }, ) ) # Sorted is stable, so within a severity each tool keeps its own order -- the rank, where it has one. rows.sort(key=lambda pair: SEVERITY_ORDER.index(pair[0])) return { "id": "critical", "title": "Most Critical Vulnerabilities", "table": { "columns": ["Source", "Severity", "CVE/CWE", "Affected File / Component", "Description"], "rows": [row for _, row in rows[:MOST_CRITICAL]], }, } def _tool_section(name: str, report: SarifReport) -> dict[str, Any]: """Every finding of one tool that was not set aside.""" kept = _kept(report) group: dict[str, Any] = {"title": f"True Issues ({len(kept)})", "cards": [_finding_card(one) for one in kept]} if not kept: group["note"] = f"Every finding {name} reported is listed under False Positives and Suppressed Findings." return {"id": _anchor(name), "title": name, "new_page": True, "groups": [group]} def _set_aside_section(tools: list[tuple[str, SarifReport]]) -> dict[str, Any] | None: """The false positives and the suppressed findings of every tool, at the end of the report.""" grouped = [(name, _set_aside(report)) for name, report in tools] grouped = [(name, findings) for name, findings in grouped if findings] if not grouped: return None groups: list[dict[str, Any]] = [ { "title": "Overview", "note": ( "The following findings were reported but set aside: classified as false positives, or " "suppressed by the tool that reported them. Each entry carries the reason it was set aside." ), "table": _set_aside_table(grouped), } ] groups += [ {"title": f"{name} ({len(findings)})", "cards": [_finding_card(one, set_aside=True) for one in findings]} for name, findings in grouped ] return { "id": "set-aside", "title": "False Positives and Suppressed Findings", "new_page": True, "groups": groups, } def _finding_card(finding: Finding, *, set_aside: bool = False) -> dict[str, Any]: """One SARIF finding.""" rows = [] # A finding a tool ruled out carries no severity worth stating. if finding.level != "none": rows.append(_row("Severity", _finding_severity(finding), kind="severity")) # Not every finding points at a line: an area an agent ruled out names it in its message instead. if finding.location is not None: rows.append(_row("Location", _where(finding), kind="mono")) rows += [ _row("Description", _text(finding.message)), _row("Reported By", ", ".join(finding.tools) or None), ] if set_aside: reason = finding.polygraph_reasoning or "; ".join( one.justification for one in finding.suppressions if one.justification ) rows.append(_row("Verdict", finding.verdict_label or "Suppressed")) rows.append(_row("Reason", reason, aside=True)) elif finding.verdict_label: rows.append(_row("Evidence", finding.verdict_label)) return {"title": f"{_stated(finding.rule_id)} — {_stated(finding.rule_name)}", "rows": rows} def _set_aside_table(grouped: list[tuple[str, list[Finding]]]) -> dict[str, Any]: """How many findings of each tool and rule were set aside.""" rows: list[dict[str, Any]] = [] for name, findings in grouped: counts = Counter(one.rule_id or UNKNOWN for one in findings) rows += [{"cells": [name, rule, count]} for rule, count in counts.most_common()] rows.append({"cells": ["Total", "", sum(len(findings) for _, findings in grouped)], "total": True}) return {"columns": ["Tool", "Rule", "Set aside"], "rows": rows} def _sca_section(sca: VdrReport) -> dict[str, Any]: """The dependency findings.""" return { "id": "sca", "title": "Dependency Vulnerabilities (SBOM / SCA)", "new_page": True, "groups": [ { "title": f"True Issues by CVEs ({sca.total})", "cards": [_vulnerability_card(one) for one in sca.vulnerabilities], } ], } def _vulnerability_card(vulnerability: Vulnerability) -> dict[str, Any]: """One dependency finding, as much of it as the report stated.""" metrics = vulnerability.metrics impact = [ f"{name} ({value})" for name, value in ( ("Confidentiality", metrics.confidentiality), ("Integrity", metrics.integrity), ("Availability", metrics.availability), ) if value ] return { "title": _stated(vulnerability.id), "rows": [ _row("Affected component", ", ".join(vulnerability.component_names) or None, kind="mono"), _row("Severity", _vdr_severity(vulnerability), kind="severity"), _row("Description", vulnerability.description), _row("CWE", ", ".join(f"{one}: {vulnerability.cwe_names[one]}" for one in vulnerability.cwes) or None), _row("CVSS", f"{vulnerability.score} / 10" if vulnerability.score is not None else None), _row("Impact", ", ".join(impact) or None), _row("Exploitation likelihood (EPSS)", _percent(metrics.epss)), _row( "Fixed in", # The report names the newest version of the package, which is not the same as the # version that fixes this advisory -- so it is labelled as what it is. next( (one.latest_version for one in vulnerability.components if one.is_outdated), None, ), kind="mono", ), _row("Known for", f"{metrics.published_days_ago} days" if metrics.published_days_ago else None), _row("Recommendation", vulnerability.recommendation), ], } # --- readings ------------------------------------------------------------------------------ def _is_set_aside(finding: Finding) -> bool: """Whether a finding was classified as a false positive or suppressed by the tool.""" return finding.false_positive or bool(finding.suppressions) def _kept(report: SarifReport) -> list[Finding]: """The findings that survived filtering.""" return [one for one in report.findings if not _is_set_aside(one)] def _set_aside(report: SarifReport) -> list[Finding]: """The findings that were filtered out or suppressed.""" return [one for one in report.findings if _is_set_aside(one)] def _text(message: str) -> str: """A message on one line. Paragraph breaks -- a title above its description -- would otherwise run together.""" return re.sub(r"\s*\n\s*\n\s*", " — ", message.strip()) def _anchor(title: str) -> str: """An id the table of contents can link to.""" return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "section" def _finding_severity(finding: Finding) -> str: """The severity band a SARIF level falls in.""" return SEVERITY_OF_LEVEL.get(finding.level, "Low") def _vdr_severity(vulnerability: Vulnerability) -> str: """The severity band a vulnerability reports, capitalised the way the report bands them.""" named = (vulnerability.severity or "").capitalize() return named if named in SEVERITY_ORDER else "Low" def _where(finding: Finding) -> str | None: """Where a finding points, as ``file:line``, every location of it.""" if finding.location is None: return None return ", ".join( f"{one.file}:{one.start_line}" if one.start_line else one.file for one in [finding.location, *finding.extra_locations] ) def _percent(share: float | None) -> str | None: """A probability as a percentage, at the precision it is worth stating.""" if share is None: return None return f"{share * 100:.2f} %" if share < 0.01 else f"{share * 100:.1f} %" def _moment(stated: Any) -> str | None: # noqa: ANN401 """An ISO timestamp as the report writes it, or nothing where it cannot be read.""" if not isinstance(stated, str): return None try: return datetime.fromisoformat(stated).strftime("%A, %B %d, %Y %H:%M:%S (%Z)").replace("()", "").strip() except ValueError: return None def _duration(seconds: Any) -> str | None: # noqa: ANN401 """A number of seconds as ``HH:MM:SS``.""" if not isinstance(seconds, (int, float)) or isinstance(seconds, bool): return None whole = int(seconds) return f"{whole // 3600:02d}:{whole % 3600 // 60:02d}:{whole % 60:02d}"