From 546d4a3e524adbc0212ad0182f819ba17c627682 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 20:32:38 +0200 Subject: [PATCH 1/2] test: specify the plugins and advisory files a report names --- tests/test_provenance.py | 232 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 tests/test_provenance.py diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 0000000..323cf11 --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,232 @@ +"""A report names what its result was produced with, besides the engine and the sources. + +Advisories decide the dependency findings and the VEX statements, and they come from +plugins and advisory files that change on their own schedule. A directory check lists +each loaded plugin, by manifest name and version, and each advisory file it read, by +path; each with the SHA-256 digest of its content. Every report format shows them, and +every finding citing an advisory names, in ``advisory_source``, the plugin or the file +that provided it. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import UTC, datetime +from pathlib import Path + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache, directory_fingerprint +from coretrace_python.cli import main +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + dump_advisories, + render_sbom, + render_vex, +) +from coretrace_python.findings import Component, Severity +from coretrace_python.reporters.json_format import render_json +from coretrace_python.reporters.sarif import render_sarif +from coretrace_python.semantic.symbols import SymbolId + +MANIFEST = ( + 'name = "vuln-data"\nversion = "2026.09.25"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = ["advisories.vuln-data"]\n\n[entrypoint]\nmodule = "vuln_data"\nclass = "VulnData"\n' +) +PLUGIN = ''' +import json +from pathlib import Path +from typing import ClassVar + +from coretrace_python.dependency import load_advisories +from coretrace_python.plugins import ModelPlugin + + +class VulnData(ModelPlugin): + name: ClassVar[str] = "vuln-data" + advisories: ClassVar = load_advisories(Path(__file__).parent / "data.json") +''' +PARSE = Advisory( + "CVE-2099-0701", + "vulnlib", + "<1.1", + "parse runs code from the document", + Severity.HIGH, + entry_points=(AdvisoryEntryPoint(SymbolId("python.vulnlib.parse"), "commit abc1234"),), + modules=("vulnlib",), +) +APP = "import vulnlib\n\ndef run():\n return vulnlib.parse(input())\n" +RULES = ("vulnerable-dependency", "reachable-vulnerability", "exploitable-vulnerability") + + +def project(tmp_path: Path) -> tuple[Path, Path]: + plugin = tmp_path / "plugins" / "vuln_data" + plugin.mkdir(parents=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "vuln_data.py").write_text(PLUGIN, encoding="utf-8") + (plugin / "data.json").write_text(dump_advisories((PARSE,)), encoding="utf-8") + root = tmp_path / "project" + root.mkdir() + (root / "requirements.txt").write_text("vulnlib==1.0\n", encoding="utf-8") + (root / "app.py").write_text(APP, encoding="utf-8") + return root, tmp_path / "plugins" + + +def analyse(root: Path, plugins: Path, **options: object) -> engine.ProjectAnalysis: + return engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins], **options) # type: ignore[arg-type] + + +def sha256(path: Path) -> str: + return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() + + +def sources(analysis: engine.ProjectAnalysis) -> dict[str, set[str]]: + found: dict[str, set[str]] = {} + for finding in analysis.findings: + if finding.rule_id in RULES: + found.setdefault(finding.rule_id, set()).add(finding.metadata["advisory_source"]) + return found + + +# --------------------------------------------------------------------------- components + + +def test_a_directory_check_names_each_plugin_and_advisory_file_with_its_digest(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + (root / "advisories.json").write_text(dump_advisories(()), encoding="utf-8") + extra = tmp_path / "feed.json" + extra.write_text(dump_advisories(()), encoding="utf-8") + + components = analyse(root, plugins, advisory_files=(extra,)).components + + plugin = plugins / "vuln_data" + assert Component("plugin", "vuln-data", "sha256:" + directory_fingerprint(plugin), "2026.09.25") in components + assert Component("advisories", "advisories.json", sha256(root / "advisories.json")) in components + assert Component("advisories", str(extra), sha256(extra)) in components + assert {c.name for c in components if c.kind == "plugin"} >= {"vuln-data", "sample-advisories"} + + +def test_a_plugin_digest_follows_its_data_and_ignores_bytecode(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + plugin = plugins / "vuln_data" + + def digest() -> str: + (component,) = [c for c in analyse(root, plugins).components if c.name == "vuln-data"] + return component.digest + + first = digest() + (plugin / "__pycache__").mkdir(exist_ok=True) + (plugin / "__pycache__" / "vuln_data.cpython-312.pyc").write_bytes(b"compiled") + assert digest() == first + (plugin / "data.json").write_text(dump_advisories((PARSE, PARSE)), encoding="utf-8") + assert digest() != first + + +def test_an_advisory_file_that_does_not_load_is_no_component(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + (root / "advisories.json").write_text("{not json", encoding="utf-8") + + analysis = analyse(root, plugins) + + assert [c for c in analysis.components if c.kind == "advisories"] == [] + assert any(f.rule_id == "syntax-error" for f in analysis.findings) + + +# --------------------------------------------------------------------------- findings + + +def test_every_advisory_finding_names_the_plugin_that_provided_its_advisory(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + + assert sources(analyse(root, plugins)) == {rule: {"vuln-data@2026.09.25"} for rule in RULES} + + +def test_an_advisory_file_that_redefines_an_advisory_is_its_source(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + (root / "advisories.json").write_text(dump_advisories((PARSE,)), encoding="utf-8") + + assert sources(analyse(root, plugins)) == {rule: {"advisories.json"} for rule in RULES} + + +def test_findings_served_from_the_cache_name_their_source_too(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + cache = ProjectCache(tmp_path / "cache") + analyse(root, plugins, cache=cache) + + second = analyse(root, plugins, cache=cache) + + assert second.reused == ("app",) + assert sources(second) == {rule: {"vuln-data@2026.09.25"} for rule in RULES} + + +# --------------------------------------------------------------------------- reports + + +def test_every_report_format_shows_the_components(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + analysis = analyse(root, plugins) + (plugin,) = [c for c in analysis.components if c.name == "vuln-data"] + report = engine.report(analysis.findings, analysis.coverage, root, components=analysis.components) + + listed = json.loads(render_json(report))["tool"]["components"] + assert {"kind": "plugin", "name": "vuln-data", "version": "2026.09.25", "digest": plugin.digest} in listed + + extensions = json.loads(render_sarif(report))["runs"][0]["tool"]["extensions"] + assert {"name": "vuln-data", "version": "2026.09.25", "properties": {"kind": "plugin", "digest": plugin.digest}} in extensions + + tools = json.loads( + render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, "0.0.0", analysis.components) + )["metadata"]["tools"]["components"] + assert { + "type": "application", + "name": "vuln-data", + "version": "2026.09.25", + "hashes": [{"alg": "SHA-256", "content": plugin.digest.removeprefix("sha256:")}], + } in tools + + def vex() -> dict[str, object]: + again = analyse(root, plugins) + return json.loads( + render_vex( + again.dependencies, + again.advisories, + again.findings, + again.coverage, + root, + engine.TOOL_NAME, + "0.0.0", + datetime(2026, 9, 25, tzinfo=UTC), + again.components, + ) + ) + + first = vex() + assert f"vuln-data 2026.09.25 ({plugin.digest})" in str(first["tooling"]) + (plugins / "vuln_data" / "data.json").write_text(dump_advisories((PARSE, PARSE)), encoding="utf-8") + assert vex()["@id"] != first["@id"] + + +def test_an_advisory_file_is_a_data_component_of_the_sbom(tmp_path: Path) -> None: + root, plugins = project(tmp_path) + (root / "advisories.json").write_text(dump_advisories((PARSE,)), encoding="utf-8") + analysis = analyse(root, plugins) + + tools = json.loads( + render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, "0.0.0", analysis.components) + )["metadata"]["tools"]["components"] + + digest = sha256(root / "advisories.json").removeprefix("sha256:") + assert {"type": "data", "name": "advisories.json", "hashes": [{"alg": "SHA-256", "content": digest}]} in tools + + +def test_the_command_line_lists_components_for_a_directory_only(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root, plugins = project(tmp_path) + + main(["--check", str(root), "--plugins", str(plugins), "--format", "json"]) + directory = json.loads(capsys.readouterr().out) + main(["--check", str(root / "app.py"), "--plugins", str(plugins), "--format", "json"]) + single = json.loads(capsys.readouterr().out) + + assert "vuln-data" in {c["name"] for c in directory["tool"]["components"]} + assert "components" not in single["tool"] From ab6f167e869a889a5b79576b712444583957c254 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 20:32:38 +0200 Subject: [PATCH 2/2] feat: name the plugins and advisory files a result was produced with A directory check lists each plugin it loaded and each advisory file it read, with the SHA-256 digest of its content, and every report format shows them: JSON tool.components, SARIF tool.extensions, the SBOM tools and the VEX tooling. Every finding about an advisory names the plugin or file it comes from in advisory_source. The cache key takes each plugin's identity from the same digest the reports show. --- CHANGELOG.md | 12 ++ docs/usage.md | 34 +++++- src/coretrace_python/cli.py | 9 +- src/coretrace_python/dependency/sbom.py | 26 +++- src/coretrace_python/dependency/vex.py | 10 +- src/coretrace_python/engine.py | 115 ++++++++++++++---- src/coretrace_python/findings/__init__.py | 4 + src/coretrace_python/findings/provenance.py | 26 ++++ src/coretrace_python/reporters/json_format.py | 20 ++- src/coretrace_python/reporters/report.py | 4 +- src/coretrace_python/reporters/sarif.py | 15 ++- 11 files changed, 230 insertions(+), 45 deletions(-) create mode 100644 src/coretrace_python/findings/provenance.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f6f9ff..74afe73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +### Reports + +- A directory check names what its result was produced with, besides the engine and the sources. It lists each plugin it loaded, by manifest name and version, and each advisory file it read, by path, each with the SHA-256 digest of its content. The formats show them in these places: + - JSON: `tool.components`; + - SARIF: `tool.extensions`; + - CycloneDX SBOM: `metadata.tools.components`, with their hashes; + - OpenVEX: the document's `tooling`, so its `@id` changes with the advisory data. + + Every finding about an advisory names the plugin (`name@version`) or advisory file it comes from, in its `advisory_source` metadata. The cache key and the reports identify a plugin by the same digest (#149). + ## 0.9.0 (2026-09-25) ### Precision diff --git a/docs/usage.md b/docs/usage.md index 02e0651..a368489 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -261,10 +261,34 @@ the directory of the checked file; a path outside it is printed as it is. `root` is the directory the paths are relative to. +A directory check also lists, in `tool.components`, what the result was produced with +besides the engine and the sources: each plugin it loaded, by manifest name and version, +and each advisory file it read, by path (relative to the root when under it). Each +component carries the SHA-256 digest of its content, so a result can be traced to the +exact advisory data it used and reproduced. A plugin's digest covers its directory as the +cache fingerprints it, bytecode and hidden files left out; a file's digest covers its +bytes. A single-file check lists no components. + +```json +"tool": { + "name": "coretrace-python-analyzer", "version": "0.10.0", + "components": [ + {"kind": "plugin", "name": "curated-advisories", "version": "2026.09.25", "digest": "sha256:…"}, + {"kind": "advisories", "name": "advisories.json", "digest": "sha256:…"} + ] +} +``` + +A finding about an advisory names, in its `advisory_source` metadata, where the advisory +comes from. For a plugin this is `name@version`, and for an advisory file its path. When +several sources define the same advisory, it names the one whose definition is used, as +the next section describes. + `--format sarif` prints a SARIF 2.1.0 log, one run with the tool, its rules and one result per finding. The root is declared once as the `SRCROOT` original URI base and every location under it is relative to that base, which is what code scanning services -need to attach results to files: +need to attach results to files. The components of a directory check are the tool's +`extensions`, their kind and digest in `properties`: ```bash coretrace-python-analyzer --check src/ --format sarif > report.sarif @@ -385,7 +409,9 @@ ignore = ["CVE-2020-1747"] ``` `--sbom PATH` writes a CycloneDX 1.5 bill of materials: one component per requirement -with its package URL, and the advisories affecting them as vulnerabilities. +with its package URL, and the advisories affecting them as vulnerabilities. Its tools +list the analyzer and the components of the check, each with its SHA-256 hash: a plugin +as an `application`, an advisory file as `data`. ```bash coretrace-python-analyzer --check src/ --sbom sbom.json --policy security/policy.toml @@ -407,7 +433,9 @@ not the code of installed packages. A framework calling the vulnerable function project's behalf would go unseen. Without a lock file, a package is never `not_affected`. The document's `@id` derives from its statements, so two runs with the same result share it; its `author` is `Unknown Author`, as OpenVEX tools write when they -cannot know it. +cannot know it. Its `tooling` names the analyzer and each component with its digest +(`coretrace-python-analyzer 0.10.0; curated-advisories 2026.09.25 (sha256:…)`), so the +`@id` changes with the advisory data a statement was decided with. ## Large projects diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index 776edea..c57d350 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -242,9 +242,12 @@ def main(argv: list[str] | None = None) -> int: findings = analysis.findings coverage = analysis.coverage suppressed = analysis.suppressed + components = analysis.components if args.sbom is not None: args.sbom.write_text( - render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, __version__), + render_sbom( + analysis.dependencies, analysis.advisories, engine.TOOL_NAME, __version__, analysis.components + ), encoding="utf-8", ) if args.vex is not None: @@ -258,6 +261,7 @@ def main(argv: list[str] | None = None) -> int: engine.TOOL_NAME, __version__, datetime.now(UTC), + analysis.components, ), encoding="utf-8", ) @@ -265,6 +269,7 @@ def main(argv: list[str] | None = None) -> int: file_analysis = engine.analyze_file(SourceManager().load_file(args.path), plugin_roots) findings, coverage = file_analysis.findings, file_analysis.coverage suppressed = file_analysis.suppressed + components = () root = args.path.resolve() if args.path.is_dir() else args.path.resolve().parent baselined = None if args.baseline is not None: @@ -273,7 +278,7 @@ def main(argv: list[str] | None = None) -> int: else: Baseline.of(findings, root).save(args.baseline) findings, baselined = (), findings - report = engine.report(findings, coverage, root, suppressed, baselined) + report = engine.report(findings, coverage, root, suppressed, baselined, components) print(render(args.format or "text", report), end="") threshold = Severity(args.fail_on).rank if args.fail_on is not None else 0 failing = any(finding.severity.rank >= threshold for finding in findings) diff --git a/src/coretrace_python/dependency/sbom.py b/src/coretrace_python/dependency/sbom.py index 65b0900..4e9be7b 100644 --- a/src/coretrace_python/dependency/sbom.py +++ b/src/coretrace_python/dependency/sbom.py @@ -3,9 +3,10 @@ from __future__ import annotations import json -from collections.abc import Iterable +from collections.abc import Iterable, Sequence from coretrace_python.dependency.graph import Advisory, DependencyGraph, Requirement +from coretrace_python.findings import PLUGIN, Component SPEC_VERSION = "1.5" @@ -33,8 +34,22 @@ def _component(requirement: Requirement) -> dict[str, object]: return component +def _tool(component: Component) -> dict[str, object]: + """A plugin as an application, an advisory file as data, each with its SHA-256.""" + + tool: dict[str, object] = {"type": "application" if component.kind == PLUGIN else "data", "name": component.name} + if component.version is not None: + tool["version"] = component.version + tool["hashes"] = [{"alg": "SHA-256", "content": component.digest.removeprefix("sha256:")}] + return tool + + def render_sbom( - dependencies: DependencyGraph, advisories: Iterable[Advisory], tool_name: str, tool_version: str + dependencies: DependencyGraph, + advisories: Iterable[Advisory], + tool_name: str, + tool_version: str, + components: Sequence[Component] = (), ) -> str: """A CycloneDX JSON document: one component per requirement, and the advisories affecting them as vulnerabilities. Deterministic for a given graph.""" @@ -57,7 +72,12 @@ def render_sbom( "specVersion": SPEC_VERSION, "version": 1, "metadata": { - "tools": {"components": [{"type": "application", "name": tool_name, "version": tool_version}]} + "tools": { + "components": [ + {"type": "application", "name": tool_name, "version": tool_version}, + *(_tool(component) for component in components), + ] + } }, "components": [_component(requirement) for requirement in requirements], "vulnerabilities": vulnerabilities, diff --git a/src/coretrace_python/dependency/vex.py b/src/coretrace_python/dependency/vex.py index 1a92734..3619fdc 100644 --- a/src/coretrace_python/dependency/vex.py +++ b/src/coretrace_python/dependency/vex.py @@ -23,7 +23,7 @@ from coretrace_python.dependency.graph import Advisory, DependencyGraph, Requirement from coretrace_python.dependency.sbom import purl -from coretrace_python.findings import Finding +from coretrace_python.findings import Component, Finding from coretrace_python.findings.coverage import Coverage CONTEXT = "https://openvex.dev/ns/v0.2.0" @@ -44,6 +44,7 @@ def render_vex( tool_name: str, tool_version: str, timestamp: datetime, + components: Sequence[Component] = (), ) -> str: """The OpenVEX document of the project at ``root``. ``evidence`` is every finding of the check, those suppressed or accepted by the policy included. The document's @@ -69,7 +70,12 @@ def render_vex( **_status(advisory, requirement, about, dependencies, coverage, root), } ) - tooling = f"{tool_name} {tool_version}" + tooling = "; ".join( + ( + f"{tool_name} {tool_version}", + *(" ".join(filter(None, (c.name, c.version, f"({c.digest})"))) for c in components), + ) + ) content = { "@context": CONTEXT, "author": AUTHOR, diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 38e5833..49066da 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -7,9 +7,10 @@ from __future__ import annotations import concurrent.futures +import hashlib import multiprocessing from collections.abc import Callable, Iterable, Mapping, Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from types import MappingProxyType from typing import Any, ClassVar @@ -55,7 +56,10 @@ correlate, ) from coretrace_python.findings import ( + ADVISORIES, FINDING_SCHEMA_VERSION, + PLUGIN, + Component, Confidence, Coverage, FileCoverage, @@ -217,6 +221,8 @@ class ProjectAnalysis: suppressed: tuple[Finding, ...] = () # Findings about an advisory the policy accepts, still evidence of what is reached. accepted: tuple[Finding, ...] = () + # The plugins and advisory files the result was produced with. + components: tuple[Component, ...] = () @dataclass(frozen=True) @@ -358,10 +364,10 @@ def analyze_project( for error in dependencies.errors: findings.append(_note("syntax-error", error, sources.add_source(error.split(":")[0], ""), 1)) advisory_paths = _advisory_paths(root, advisory_files) - file_advisories: list[Advisory] = [] + file_advisories: list[tuple[Path, tuple[Advisory, ...]]] = [] for path in advisory_paths: try: - file_advisories.extend(load_advisories(path)) + file_advisories.append((path, load_advisories(path))) except AdvisoryFileError as error: findings.append(_note("syntax-error", str(error), sources.add_source(str(path), ""), 1)) policy = Policy() @@ -392,9 +398,8 @@ def analyze_project( probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("", ""))) registry = load_plugins(plugin_roots, probe) all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *plugins) - advisories = _merge_advisories( - (a for plugin in all_plugins for a in plugin.advisories), file_advisories - ) + components = _components(registry, root, (path for path, _ in file_advisories)) + advisories, origins = _merge_advisories(_contributions(registry, plugins, root, file_advisories)) affected = affected_symbols(dependencies, advisories) models = plugin_models(all_plugins, root).extended(*advisory_sinks(affected)) for manager in managers.values(): @@ -425,7 +430,7 @@ def analyze_project( manager.provide(EscapedTemplates, escaped) manager.provide(ClearingAnalysis, clearing) - configuration = _configuration_key(registry, plugins, models, advisories, dependencies, routes, escaped) + configuration = _configuration_key(components, plugins, models, advisories, dependencies, routes, escaped) keys = module_keys( graph, {name: fingerprint(configuration, str(files[name].source_id), name, files[name].text) for name in analysable}, @@ -511,6 +516,7 @@ def analyze_project( for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): findings = list(apply_refinement(plugin, findings, plugin.refine(context, tuple(findings)))) + findings = [_sourced(finding, origins) for finding in findings] accepted = tuple(f for f in findings if policy.accepts(f)) kept, suppressed = partition(apply_policy(policy, findings), _text_of(sources)) return ProjectAnalysis( @@ -524,20 +530,78 @@ def analyze_project( Coverage(tuple(sorted(coverage, key=lambda c: c.path))), suppressed, accepted, + components, + ) + + +Origins = Mapping[tuple[str, str], str] + + +def _merge_advisories(contributions: Iterable[tuple[str, Iterable[Advisory]]]) -> tuple[tuple[Advisory, ...], Origins]: + """One advisory per identifier and package, with the contributor it comes from: a + later contributor replaces an earlier one, so a curated plugin refines a bundled + sample and a local file, the project's own feed, wins over every plugin.""" + + merged: dict[tuple[str, str], tuple[str, Advisory]] = {} + for source, advisories in contributions: + for advisory in advisories: + merged[(advisory.id, advisory.package)] = (source, advisory) + return ( + tuple(advisory for _, advisory in merged.values()), + MappingProxyType({key: source for key, (source, _) in merged.items()}), ) -def _merge_advisories( - from_plugins: Iterable[Advisory], from_files: Iterable[Advisory] -) -> tuple[Advisory, ...]: - """One advisory per identifier and package: a later contributor replaces an earlier - one, so a curated plugin refines a bundled sample and a local file, the project's - own feed, wins over every plugin.""" +def _contributions( + registry: PluginRegistry, + plugins: Sequence[Plugin], + root: Path, + files: Iterable[tuple[Path, tuple[Advisory, ...]]], +) -> list[tuple[str, Iterable[Advisory]]]: + """The advisories of each plugin, then of each advisory file, each under the name a + finding cites it by: ``name@version`` for a loaded plugin, the path for a file.""" + + return [ + *((f"{loaded.manifest.name}@{loaded.manifest.version}", loaded.plugin.advisories) for loaded in registry), + *((plugin.name, plugin.advisories) for plugin in plugins), + *((_located(path, root), advisories) for path, advisories in files), + ] - merged: dict[tuple[str, str], Advisory] = {} - for advisory in (*from_plugins, *from_files): - merged[(advisory.id, advisory.package)] = advisory - return tuple(merged.values()) + +def _components(registry: PluginRegistry, root: Path, files: Iterable[Path]) -> tuple[Component, ...]: + """The plugins and advisory files a project result is produced with, each with the + digest of its content: the plugin directory as the cache fingerprints it, the file's + bytes.""" + + return ( + *( + Component(PLUGIN, loaded.manifest.name, f"sha256:{directory_fingerprint(loaded.directory)}", loaded.manifest.version) + for loaded in registry + ), + *( + Component(ADVISORIES, _located(path, root), f"sha256:{hashlib.sha256(path.read_bytes()).hexdigest()}") + for path in files + ), + ) + + +def _located(path: Path, root: Path) -> str: + """``path`` relative to ``root``, POSIX style, when it lies under it.""" + + try: + return path.resolve().relative_to(root.resolve()).as_posix() + except ValueError: + return str(path) + + +def _sourced(finding: Finding, origins: Origins) -> Finding: + """``finding`` with the plugin or advisory file its advisory comes from, when it + cites one.""" + + origin = origins.get((finding.metadata.get("advisory", ""), finding.metadata.get("package", ""))) + if origin is None: + return finding + return replace(finding, metadata={**finding.metadata, "advisory_source": origin}) def _advisory_paths(root: Path, advisory_files: Sequence[Path]) -> tuple[Path, ...]: @@ -623,15 +687,13 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]: registry = load_plugins(batch.plugin_roots, next(iter(managers.values()))) all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *batch.plugins) dependencies = resolve_dependencies(batch.root, sources) - file_advisories: list[Advisory] = [] + file_advisories: list[tuple[Path, tuple[Advisory, ...]]] = [] for path in batch.advisory_paths: try: - file_advisories.extend(load_advisories(path)) + file_advisories.append((path, load_advisories(path))) except AdvisoryFileError: continue - advisories = _merge_advisories( - (a for plugin in all_plugins for a in plugin.advisories), file_advisories - ) + advisories, _ = _merge_advisories(_contributions(registry, batch.plugins, batch.root, file_advisories)) affected = affected_symbols(dependencies, advisories) models = plugin_models(all_plugins, batch.root).extended(*advisory_sinks(affected)) routes = _decode_routes(batch.routes) @@ -663,7 +725,7 @@ def _decode_routes(data: tuple[tuple[str, str, str, int], ...]) -> dict[SymbolId def _configuration_key( - registry: PluginRegistry, + components: Sequence[Component], plugins: Sequence[Plugin], models: ModelTable, advisories: tuple[Advisory, ...], @@ -679,10 +741,7 @@ def _configuration_key( str(PYIR_SCHEMA_VERSION), str(FINDING_SCHEMA_VERSION), str(PLUGIN_API_VERSION), - *( - f"{loaded.manifest.name}={loaded.manifest.version}:{directory_fingerprint(loaded.directory)}" - for loaded in registry - ), + *(f"{c.name}={c.version}:{c.digest}" for c in components if c.kind == PLUGIN), *(f"{type(p).__module__}.{type(p).__qualname__}" for p in plugins), repr(models), repr(advisories), @@ -776,6 +835,7 @@ def report( root: Path | None = None, suppressed: Sequence[Finding] = (), baselined: Sequence[Finding] | None = None, + components: Sequence[Component] = (), ) -> Report: """The report of a check; paths under ``root`` are rendered relative to it. ``baselined`` is the findings a baseline accounts for, ``None`` when none applied.""" @@ -789,4 +849,5 @@ def report( tuple(suppressed), tuple(baselined or ()), baselined is not None, + tuple(components), ) diff --git a/src/coretrace_python/findings/__init__.py b/src/coretrace_python/findings/__init__.py index 62b859e..c263417 100644 --- a/src/coretrace_python/findings/__init__.py +++ b/src/coretrace_python/findings/__init__.py @@ -2,9 +2,13 @@ from coretrace_python.findings.coverage import Coverage, FileCoverage from coretrace_python.findings.model import FINDING_SCHEMA_VERSION, Confidence, Finding, Severity +from coretrace_python.findings.provenance import ADVISORIES, PLUGIN, Component __all__ = [ + "ADVISORIES", "FINDING_SCHEMA_VERSION", + "PLUGIN", + "Component", "Confidence", "Coverage", "FileCoverage", diff --git a/src/coretrace_python/findings/provenance.py b/src/coretrace_python/findings/provenance.py new file mode 100644 index 0000000..adedd85 --- /dev/null +++ b/src/coretrace_python/findings/provenance.py @@ -0,0 +1,26 @@ +"""What a result was produced with, besides the engine and the analysed sources. + +Advisories decide the dependency findings and the VEX statements, and they come from +plugins and advisory files that change on their own schedule. A report names each of +them with the digest of its content, so a result can be traced to its data and +reproduced. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +PLUGIN = "plugin" +ADVISORIES = "advisories" + + +@dataclass(frozen=True) +class Component: + """A loaded plugin, by its manifest ``name`` and ``version``, or an advisory file, by + its path as ``name``. ``digest`` is ``sha256:`` and the hex digest of its content: + the plugin directory as the cache fingerprints it, or the file's bytes.""" + + kind: str + name: str + digest: str + version: str | None = None diff --git a/src/coretrace_python/reporters/json_format.py b/src/coretrace_python/reporters/json_format.py index a928458..1042add 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/src/coretrace_python/reporters/json_format.py @@ -4,7 +4,7 @@ import json -from coretrace_python.findings import FINDING_SCHEMA_VERSION, Finding +from coretrace_python.findings import FINDING_SCHEMA_VERSION, Component, Finding from coretrace_python.reporters.report import Report @@ -28,11 +28,21 @@ def finding_record(finding: Finding, report: Report | None = None) -> dict[str, } +def component_record(component: Component) -> dict[str, str]: + """A plugin or advisory file the result was produced with; a file has no version.""" + + record = {"kind": component.kind, "name": component.name} + if component.version is not None: + record["version"] = component.version + record["digest"] = component.digest + return record + + def render_json(report: Report) -> str: - document: dict[str, object] = { - "schema_version": FINDING_SCHEMA_VERSION, - "tool": {"name": report.tool_name, "version": report.tool_version}, - } + tool: dict[str, object] = {"name": report.tool_name, "version": report.tool_version} + if report.components: + tool["components"] = [component_record(component) for component in report.components] + document: dict[str, object] = {"schema_version": FINDING_SCHEMA_VERSION, "tool": tool} if report.root is not None: document["root"] = str(report.root) document["findings"] = [finding_record(finding, report) for finding in report.findings] diff --git a/src/coretrace_python/reporters/report.py b/src/coretrace_python/reporters/report.py index 90f4a81..95bb5af 100644 --- a/src/coretrace_python/reporters/report.py +++ b/src/coretrace_python/reporters/report.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from pathlib import Path, PurePosixPath -from coretrace_python.findings import Coverage, Finding +from coretrace_python.findings import Component, Coverage, Finding def _order(finding: Finding) -> tuple[str, int, int, str]: @@ -27,6 +27,8 @@ class Report: baselined: tuple[Finding, ...] = () # Whether a baseline was applied, so new findings can be marked as such. with_baseline: bool = False + # The plugins and advisory files the result was produced with. + components: tuple[Component, ...] = () def __post_init__(self) -> None: for name in ("findings", "suppressed", "baselined"): diff --git a/src/coretrace_python/reporters/sarif.py b/src/coretrace_python/reporters/sarif.py index c752da5..9139a5f 100644 --- a/src/coretrace_python/reporters/sarif.py +++ b/src/coretrace_python/reporters/sarif.py @@ -4,7 +4,7 @@ import json -from coretrace_python.findings import Finding, Severity +from coretrace_python.findings import Component, Finding, Severity from coretrace_python.reporters.report import Report SARIF_VERSION = "2.1.0" @@ -59,6 +59,16 @@ def _result( return result +def _extension(component: Component) -> dict[str, object]: + """A plugin or advisory file as a SARIF tool extension, its kind and digest as properties.""" + + extension: dict[str, object] = {"name": component.name} + if component.version is not None: + extension["version"] = component.version + extension["properties"] = {"kind": component.kind, "digest": component.digest} + return extension + + def render_sarif(report: Report) -> str: rule_ids: list[str] = [] for finding in (*report.findings, *report.suppressed, *report.baselined): @@ -78,7 +88,8 @@ def render_sarif(report: Report) -> str: {"id": rule_id, "shortDescription": {"text": rule_id}} for rule_id in rule_ids ], - } + }, + **({"extensions": [_extension(c) for c in report.components]} if report.components else {}), }, "results": [ *(_result(report, f, rule_ids.index(f.rule_id), False, new) for f in report.findings),