From 2d526b14a05e26ea2cbf208b87e6d898e9cc18d6 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 19:25:28 +0200 Subject: [PATCH 1/2] test: specify OpenVEX statements and the packages a lock file shows requiring each --- tests/test_dependencies.py | 80 +++++++++++ tests/test_vex.py | 284 +++++++++++++++++++++++++++++++++++++ 2 files changed, 364 insertions(+) create mode 100644 tests/test_vex.py diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py index 3b6b93d..8b44950 100644 --- a/tests/test_dependencies.py +++ b/tests/test_dependencies.py @@ -147,6 +147,86 @@ def test_graphs_merge_and_locks_pin_declared_requirements() -> None: assert merged.names == ("pyyaml",) +UV_LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "flask" }, { name = "pyyaml" }] + +[package.dev-dependencies] +dev = [{ name = "pytest" }] + +[[package]] +name = "flask" +version = "3.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [{ name = "Werkzeug" }, { name = "jinja2", marker = "python_version >= '3.8'" }] + +[package.optional-dependencies] +dotenv = [{ name = "python-dotenv" }] + +[[package]] +name = "werkzeug" +version = "3.0.1" +source = { registry = "https://pypi.org/simple" } + +[[package]] +name = "pyyaml" +version = "6.0.1" +source = { registry = "https://pypi.org/simple" } + +[[package]] +name = "pytest" +version = "8.0.0" +source = { registry = "https://pypi.org/simple" } +""" + + +def test_lockfiles_tell_which_other_packages_require_a_package() -> None: + uv = parse_dependencies(source("uv.lock", UV_LOCK)) + poetry = parse_dependencies( + source( + "poetry.lock", + '[[package]]\nname = "flask"\nversion = "3.0.0"\n\n[package.dependencies]\nWerkzeug = ">=3.0.0"\n' + 'python-dotenv = {version = "*", optional = true}\n\n[[package]]\nname = "werkzeug"\nversion = "3.0.1"\n' + '\n[[package]]\nname = "pyyaml"\nversion = "6.0.1"\n', + ) + ) + + for graph in (uv, poetry): + assert graph.required_by("werkzeug") == frozenset({"flask"}) + assert graph.required_by("python-dotenv") == frozenset({"flask"}) + assert graph.required_by("pyyaml") == frozenset() + assert graph.required_by("requests") is None + # The project's own packages are not other packages: only its code uses pyyaml. + assert uv.required_by("jinja2") == frozenset({"flask"}) + assert uv.required_by("pytest") == frozenset() + assert parse_dependencies(source("requirements.txt", "pyyaml==6.0.1\n")).required_by("pyyaml") is None + + +def test_a_package_outside_the_project_is_another_package_even_editable() -> None: + lock = UV_LOCK.replace('[[package]]\nname = "pytest"', '[[package]]\nname = "shared"\nversion = "1.0"\n' + 'source = { editable = "../shared" }\ndependencies = [{ name = "pyyaml" }]\n\n' + '[[package]]\nname = "pytest"') + + assert parse_dependencies(source("uv.lock", lock)).required_by("pyyaml") == frozenset({"shared"}) + + +def test_merged_graphs_keep_what_every_lock_file_says() -> None: + declared = parse_dependencies(source("requirements.txt", "werkzeug==3.0.1\n")) + uv = parse_dependencies(source("uv.lock", UV_LOCK)) + poetry = parse_dependencies( + source("poetry.lock", '[[package]]\nname = "django-extra"\nversion = "1.0"\n\n[package.dependencies]\nwerkzeug = "*"\n') + ) + + merged = declared.merge(uv).merge(poetry) + + assert merged.required_by("werkzeug") == frozenset({"flask", "django-extra"}) + assert declared.merge(uv).required_by("werkzeug") == uv.merge(declared).required_by("werkzeug") == frozenset({"flask"}) + + # --------------------------------------------------------------------------- versions diff --git a/tests/test_vex.py b/tests/test_vex.py new file mode 100644 index 0000000..b554bc8 --- /dev/null +++ b/tests/test_vex.py @@ -0,0 +1,284 @@ +"""OpenVEX statements from the evidence of a directory check. + +``--vex PATH`` writes one statement per advisory affecting a requirement. The project is +``affected`` when its code reaches the vulnerability, reachable or exploitable, whatever +the policy or a suppression does with the finding. It is ``not_affected`` +(``vulnerable_code_not_in_execute_path``) only when a curated advisory's entry points +are not reached, every file and function was analysed, and a lock file shows that no +other package requires the vulnerable one: the engine does not analyse the code of +installed packages. Anything else is ``under_investigation``, and the statement says +why. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + Condition, + dump_advisories, + render_vex, +) +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId + +PLUGINS = engine.BUNDLED_PLUGINS +NOW = datetime(2026, 9, 24, 12, 0, tzinfo=UTC) + +CURATED = Advisory( + "CVE-2099-0301", + "vulnlib", + "<1.1", + "parse runs code from the document in unsafe mode", + Severity.CRITICAL, + aliases=("GHSA-aaaa-bbbb-cccc",), + entry_points=( + AdvisoryEntryPoint( + SymbolId("python.vulnlib.parse"), + "parse builds objects from tags in unsafe mode (commit abc1234)", + ( + Condition( + "argument", + "mode is UNSAFE, the default", + argument="mode", + values=("python.vulnlib.UNSAFE",), + position=1, + default=True, + ), + ), + ), + ), + modules=("vulnlib",), +) +UNCURATED = Advisory("CVE-2099-0302", "vulnlib", "<1.1", "a crafted document exhausts memory", Severity.MEDIUM) + +PROJECT_LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "vulnlib" }] + +[[package]] +name = "vulnlib" +version = "1.0" +source = { registry = "https://pypi.org/simple" } +""" +FRAMEWORK_LOCK = PROJECT_LOCK + """ +[[package]] +name = "framework" +version = "2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [{ name = "vulnlib" }] +""" + + +def vex(root: Path, app: str, files: dict[str, str] | None = None) -> dict[str, Any]: + for relative, text in {"uv.lock": PROJECT_LOCK, "app.py": app, **(files or {})}.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((CURATED, UNCURATED)), encoding="utf-8") + analysis = engine.analyze_project(root, [PLUGINS]) + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.6.0", NOW + ) + return json.loads(document) # type: ignore[no-any-return] + + +def statement(document: dict[str, Any], name: str) -> dict[str, Any]: + return next(s for s in document["statements"] if s["vulnerability"]["name"] == name) + + +CALLED = "import vulnlib\n\ndef load(text):\n return vulnlib.parse(text)\n" +IMPORTED = "import vulnlib\n\nMODE = vulnlib.SAFE\n" + + +# --------------------------------------------------------------------------- affected + + +def test_a_reached_entry_point_is_affected_with_where_and_what_to_do(tmp_path: Path) -> None: + found = statement(vex(tmp_path, CALLED), "CVE-2099-0301") + + assert found["status"] == "affected" + assert found["action_statement"] == "Update vulnlib to a version outside the vulnerable range <1.1." + assert found["status_notes"] == "Reached by the project's code at app.py:4 python.vulnlib.parse (reachable)." + + +def test_attacker_input_reaching_it_is_affected_and_said_so(tmp_path: Path) -> None: + app = "import vulnlib\n\ndef load():\n return vulnlib.parse(input())\n" + + found = statement(vex(tmp_path, app), "CVE-2099-0301") + + assert found["status"] == "affected" + assert found["status_notes"] == "Reached by the project's code at app.py:4 python.vulnlib.parse (exploitable)." + + +@pytest.mark.parametrize( + "app, files", + [ + (CALLED, {"coretrace-policy.toml": '[advisories]\nignore = ["CVE-2099-0301"]\n'}), + (CALLED.replace("(text)\n", "(text) # coretrace: ignore\n"), {}), + ], + ids=["accepted-by-the-policy", "suppressed-inline"], +) +def test_a_finding_the_project_accepts_or_silences_is_still_affected( + tmp_path: Path, app: str, files: dict[str, str] +) -> None: + assert statement(vex(tmp_path, app, files), "CVE-2099-0301")["status"] == "affected" + + +# --------------------------------------------------------------------------- not_affected + + +def test_a_curated_advisory_nothing_reaches_is_not_affected(tmp_path: Path) -> None: + found = statement(vex(tmp_path, IMPORTED), "CVE-2099-0301") + + assert found["status"] == "not_affected" + assert found["justification"] == "vulnerable_code_not_in_execute_path" + assert found["impact_statement"] == ( + "No code of the project reaches the entry points of CVE-2099-0301 (python.vulnlib.parse), " + "and no other locked package requires vulnlib." + ) + + +def test_calls_ruled_out_by_their_arguments_are_named_in_the_impact_statement(tmp_path: Path) -> None: + app = "import vulnlib\n\ndef load(text):\n return vulnlib.parse(text, vulnlib.SAFE)\n" + + found = statement(vex(tmp_path, app), "CVE-2099-0301") + + assert found["status"] == "not_affected" + assert found["impact_statement"].endswith( + " Calls ruled out by their arguments: app:4 python.vulnlib.parse(mode=python.vulnlib.SAFE)." + ) + + +# --------------------------------------------------------------------------- under_investigation + + +def test_an_advisory_naming_no_entry_point_stays_under_investigation(tmp_path: Path) -> None: + found = statement(vex(tmp_path, IMPORTED), "CVE-2099-0302") + + assert found["status"] == "under_investigation" + assert found["status_notes"] == ( + "The advisory names no entry point, so whether the project reaches the vulnerable code is not known." + ) + + +def test_a_package_another_locked_package_requires_stays_under_investigation(tmp_path: Path) -> None: + found = statement(vex(tmp_path, IMPORTED, {"uv.lock": FRAMEWORK_LOCK}), "CVE-2099-0301") + + assert found["status"] == "under_investigation" + assert found["status_notes"] == ( + "No code of the project reaches the entry points of CVE-2099-0301 (python.vulnlib.parse), " + "but vulnlib is required by framework, whose code is not analysed." + ) + + +def test_without_a_lock_file_it_stays_under_investigation(tmp_path: Path) -> None: + files = {"uv.lock": "version = 1\n", "requirements.txt": "vulnlib==1.0\n"} + + found = statement(vex(tmp_path, IMPORTED, files), "CVE-2099-0301") + + assert found["status"] == "under_investigation" + assert found["status_notes"].endswith( + "but no lock file shows which installed packages require vulnlib, and their code is not analysed." + ) + + +def test_an_incomplete_analysis_stays_under_investigation(tmp_path: Path) -> None: + found = statement(vex(tmp_path, IMPORTED, {"broken.py": "def f(:\n"}), "CVE-2099-0301") + + assert found["status"] == "under_investigation" + assert found["status_notes"] == ( + "No analysed code reaches the entry points of CVE-2099-0301 (python.vulnlib.parse), " + "but broken.py could not be fully analysed." + ) + + +# --------------------------------------------------------------------------- document + + +def test_the_document_is_openvex_about_the_project_and_its_dependency(tmp_path: Path) -> None: + document = vex(tmp_path, CALLED) + + assert document["@context"] == "https://openvex.dev/ns/v0.2.0" + assert document["@id"].startswith("https://openvex.dev/docs/public/vex-") + assert document["author"] == "Unknown Author" + assert document["timestamp"] == "2026-09-24T12:00:00Z" + assert document["version"] == 1 + assert document["tooling"] == "coretrace 0.6.0" + assert [s["vulnerability"]["name"] for s in document["statements"]] == ["CVE-2099-0301", "CVE-2099-0302"] + found = statement(document, "CVE-2099-0301") + assert found["vulnerability"] == { + "name": "CVE-2099-0301", + "description": "parse runs code from the document in unsafe mode", + "aliases": ["GHSA-aaaa-bbbb-cccc"], + } + assert found["products"] == [ + {"@id": f"pkg:generic/{tmp_path.name}", "subcomponents": [{"@id": "pkg:pypi/vulnlib@1.0"}]} + ] + + +def test_the_document_identifier_depends_on_the_statements_not_the_time(tmp_path: Path) -> None: + root = tmp_path / "src" + root.mkdir() + (root / "uv.lock").write_text(PROJECT_LOCK, encoding="utf-8") + (root / "app.py").write_text(CALLED, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((CURATED,)), encoding="utf-8") + analysis = engine.analyze_project(root, [PLUGINS]) + + def identifier(evidence: tuple[Any, ...], when: datetime) -> str: + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.6.0", when + ) + return json.loads(document)["@id"] # type: ignore[no-any-return] + + later = datetime(2027, 1, 1, tzinfo=UTC) + assert identifier(analysis.findings, NOW) == identifier(analysis.findings, later) + assert identifier(analysis.findings, NOW) != identifier((), NOW) + + +def test_advisories_affecting_no_requirement_have_no_statement(tmp_path: Path) -> None: + safe = PROJECT_LOCK.replace('version = "1.0"', 'version = "1.1"') + + assert vex(tmp_path, CALLED, {"uv.lock": safe})["statements"] == [] + + +# --------------------------------------------------------------------------- CLI + + +def test_check_writes_the_vex_document_next_to_the_report(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + root = tmp_path / "src" + root.mkdir() + (root / "uv.lock").write_text(PROJECT_LOCK, encoding="utf-8") + (root / "app.py").write_text(CALLED, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((CURATED,)), encoding="utf-8") + path = tmp_path / "vex.json" + + code = main(["--check", str(root), "--plugins", str(PLUGINS), "--vex", str(path)]) + + assert code == 1 + assert "reachable-vulnerability" in capsys.readouterr().out + document = json.loads(path.read_text(encoding="utf-8")) + assert [(s["vulnerability"]["name"], s["status"]) for s in document["statements"]] == [("CVE-2099-0301", "affected")] + assert document["timestamp"].endswith("Z") + + +def test_vex_option_requires_a_directory_check(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + source = tmp_path / "x.py" + source.write_text("", encoding="utf-8") + + assert main(["--check", "--vex", str(tmp_path / "vex.json"), str(source)]) == 2 + assert "--vex" in capsys.readouterr().err From 7082eae1927e60f09787bc78db082d48387d3f65 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 19:25:28 +0200 Subject: [PATCH 2/2] feat: write an OpenVEX document justified by reachability --- CHANGELOG.md | 5 + docs/usage.md | 21 ++- src/coretrace_python/cli.py | 34 +++- src/coretrace_python/dependency/__init__.py | 2 + src/coretrace_python/dependency/graph.py | 78 ++++++++- src/coretrace_python/dependency/policy.py | 8 +- src/coretrace_python/dependency/vex.py | 182 ++++++++++++++++++++ src/coretrace_python/engine.py | 4 + 8 files changed, 322 insertions(+), 12 deletions(-) create mode 100644 src/coretrace_python/dependency/vex.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f9f334..2fdb2c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,11 @@ - A sink call can be made safe by one of its arguments: a `SafeArgument` model names the argument and the values that take kinds off the sink, only when the call gives one explicitly. `yaml.load` with `SafeLoader`, `BaseLoader` or their C versions, under every spelling, is no longer an insecure deserialization; an absent loader, another loader, a variable or unpacked arguments still are (#121). +### Reports + +- `--vex PATH` writes an OpenVEX document with one statement per advisory affecting a requirement. A statement is `affected` when the project's code reaches the vulnerability, including findings the policy accepts or a comment suppresses. It is `not_affected` (`vulnerable_code_not_in_execute_path`) only for an advisory with entry points that nothing reaches, in a complete analysis, and when the lock file shows no other package requiring the vulnerable one. Anything else is `under_investigation`, with the reason (#97). +- A lock file records which packages require which. `DependencyGraph.required_by(name)` gives the other packages requiring a package; the project's own packages are not counted. `ProjectAnalysis.accepted` keeps the findings of advisories the policy accepts (#97). + ## 0.5.0 (2026-09-24) ### Dependencies diff --git a/docs/usage.md b/docs/usage.md index 67dda5d..3f5c5bb 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -50,6 +50,7 @@ coretrace-python-analyzer [--check | --emit-ir [--ssa]] [options] [path] | `--cache DIR` | Keep per-module results under `DIR` and reuse them for unchanged modules. | | `--jobs N` | Analyse independent modules in `N` processes. | | `--sbom PATH` | Write a CycloneDX bill of materials of the dependencies to `PATH`. | +| `--vex PATH` | Write an OpenVEX document to `PATH`: whether each advisory affecting a requirement affects the project. | | `--advisories FILE` | Read a local advisory file in addition to `advisories.json` at the root. Repeatable. | | `--policy FILE` | Apply this dependency policy instead of `coretrace-policy.toml` at the root. | | `--import-advisories SRC OUT` | Convert an OSV dump into the local advisory file `OUT`, then exit. | @@ -59,7 +60,7 @@ coretrace-python-analyzer [--check | --emit-ir [--ssa]] [options] [path] | `--ssa` | With `--emit-ir`, print the static single assignment form. | | `--help` | Show the options and exit. | -`--cache`, `--jobs`, `--sbom`, `--advisories` and `--policy` apply to a directory check. +`--cache`, `--jobs`, `--sbom`, `--vex`, `--advisories` and `--policy` apply to a directory check. ## What is analysed @@ -332,6 +333,24 @@ with its package URL, and the advisories affecting them as vulnerabilities. coretrace-python-analyzer --check src/ --sbom sbom.json --policy security/policy.toml ``` +`--vex PATH` writes an OpenVEX 0.2.0 document with one statement per advisory affecting +a requirement. In each statement, the required package is a subcomponent of the project, +identified as `pkg:generic/`. The status follows from the evidence of the +check, including findings the policy accepts or a comment suppresses: + +| Status | When | +|---|---| +| `affected` | The project's code reaches the vulnerable code. The notes give each place and its level, `reachable` or `exploitable`. | +| `not_affected` | Justified as `vulnerable_code_not_in_execute_path`, only when all of these hold: the advisory names entry points, no code of the project reaches them, every file and function was analysed, and a lock file (`uv.lock` or `poetry.lock`) shows that no other package requires the vulnerable one. The impact statement names the entry points and any calls ruled out by their arguments. | +| `under_investigation` | Anything else. The notes say which condition failed. | + +`not_affected` needs the lock-file check because the analyzer reads the project's code, +not the code of installed packages. A framework calling the vulnerable function on the +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. + ## Large projects `--cache DIR` keeps the results of a directory check on disk, one entry per module, diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index d3aefa3..bac85df 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -3,13 +3,20 @@ import argparse import sys import zipfile +from datetime import UTC, datetime from pathlib import Path from coretrace_python import __version__, engine from coretrace_python.analysis import AnalysisError from coretrace_python.cache import ProjectCache from coretrace_python.cfg import CFGError -from coretrace_python.dependency import dump_advisories, import_osv, read_osv, render_sbom +from coretrace_python.dependency import ( + dump_advisories, + import_osv, + read_osv, + render_sbom, + render_vex, +) from coretrace_python.findings import Severity from coretrace_python.findings.baseline import Baseline, BaselineError from coretrace_python.frontend import HIRBuildError, ParseError, build_hir @@ -106,6 +113,14 @@ def build_parser() -> argparse.ArgumentParser: metavar="PATH", help="with --check on a directory, write a CycloneDX bill of materials to PATH", ) + parser.add_argument( + "--vex", + type=Path, + default=None, + metavar="PATH", + help="with --check on a directory, write to PATH an OpenVEX document saying whether " + "each advisory affecting a requirement affects the project", + ) parser.add_argument( "--advisories", action="append", @@ -182,6 +197,9 @@ def main(argv: list[str] | None = None) -> int: if args.sbom is not None and not (args.check and args.path.is_dir()): print("error: --sbom only applies to --check on a directory", file=sys.stderr) return EXIT_ERROR + if args.vex is not None and not (args.check and args.path.is_dir()): + print("error: --vex only applies to --check on a directory", file=sys.stderr) + return EXIT_ERROR if args.advisories and not (args.check and args.path.is_dir()): print("error: --advisories only applies to --check on a directory", file=sys.stderr) return EXIT_ERROR @@ -227,6 +245,20 @@ def main(argv: list[str] | None = None) -> int: render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, __version__), encoding="utf-8", ) + if args.vex is not None: + args.vex.write_text( + render_vex( + analysis.dependencies, + analysis.advisories, + (*analysis.findings, *analysis.suppressed, *analysis.accepted), + analysis.coverage, + args.path, + engine.TOOL_NAME, + __version__, + datetime.now(UTC), + ), + encoding="utf-8", + ) else: file_analysis = engine.analyze_file(SourceManager().load_file(args.path), plugin_roots) findings, coverage = file_analysis.findings, file_analysis.coverage diff --git a/src/coretrace_python/dependency/__init__.py b/src/coretrace_python/dependency/__init__.py index e66aaf9..f399fb9 100644 --- a/src/coretrace_python/dependency/__init__.py +++ b/src/coretrace_python/dependency/__init__.py @@ -22,6 +22,7 @@ ) from coretrace_python.dependency.policy import POLICY_FILE, Policy, apply_policy, load_policy from coretrace_python.dependency.sbom import render_sbom +from coretrace_python.dependency.vex import render_vex __all__ = [ "ADVISORY_FILE", @@ -45,4 +46,5 @@ "parse_dependencies", "read_osv", "render_sbom", + "render_vex", ] diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 3450ec2..12d010c 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -1,7 +1,8 @@ """Dependency resolution (architecture §26). Requirements declared in ``requirements.txt``, ``pyproject.toml`` (PEP 621 and Poetry) -and pinned in ``poetry.lock`` or ``uv.lock`` become a ``DependencyGraph``. Versions are +and pinned in ``poetry.lock`` or ``uv.lock`` become a ``DependencyGraph``; a lock file +also tells which other packages require each package. Versions are compared with a small PEP 440 subset (``==``, ``!=``, ``<``, ``<=``, ``>``, ``>=``, ``~=`` and Poetry's ``^``), enough to decide whether a requirement may allow a version an advisory marks as vulnerable. Nothing is downloaded. @@ -11,9 +12,9 @@ import re import tomllib -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Iterator, Mapping from dataclasses import dataclass -from pathlib import PurePath +from pathlib import PurePath, PurePosixPath from types import MappingProxyType from typing import Any, ClassVar @@ -219,9 +220,15 @@ def imported_by(self, symbols: Iterable[SymbolId]) -> bool: class DependencyGraph: - def __init__(self, requirements: Mapping[str, Requirement] | None = None, errors: tuple[str, ...] = ()) -> None: + def __init__( + self, + requirements: Mapping[str, Requirement] | None = None, + errors: tuple[str, ...] = (), + dependents: Mapping[str, frozenset[str]] | None = None, + ) -> None: self._requirements = MappingProxyType(dict(sorted((requirements or {}).items()))) self.errors = errors + self._dependents = MappingProxyType(dict(dependents or {})) @property def names(self) -> tuple[str, ...]: @@ -234,6 +241,12 @@ def requirements(self) -> tuple[Requirement, ...]: def requirement(self, name: str) -> Requirement | None: return self._requirements.get(normalize(name)) + def required_by(self, name: str) -> frozenset[str] | None: + """The other packages a lock file shows requiring ``name`` — an empty set when + only the project does — or None when no lock file mentions it.""" + + return self._dependents.get(normalize(name)) + def merge(self, other: DependencyGraph) -> DependencyGraph: merged = dict(self._requirements) for name, requirement in other._requirements.items(): @@ -248,7 +261,10 @@ def merge(self, other: DependencyGraph) -> DependencyGraph: requirement.pinned or current.pinned, current.optional and requirement.optional, ) - return DependencyGraph(merged, self.errors + other.errors) + dependents = dict(self._dependents) + for name, found in other._dependents.items(): + dependents[name] = dependents.get(name, frozenset()) | found + return DependencyGraph(merged, self.errors + other.errors, dependents) def parse_dependencies(source: SourceFile) -> DependencyGraph: @@ -260,7 +276,7 @@ def parse_dependencies(source: SourceFile) -> DependencyGraph: if name == "pyproject.toml": return _parse_toml(source, _pyproject_requirements) if name in ("poetry.lock", "uv.lock"): - return _parse_toml(source, _lock_requirements) + return _parse_toml(source, _lock_requirements, _lock_dependents) return DependencyGraph() @@ -276,7 +292,7 @@ def _parse_requirements_txt(source: SourceFile) -> DependencyGraph: return DependencyGraph(found) -def _parse_toml(source: SourceFile, extract: Any) -> DependencyGraph: +def _parse_toml(source: SourceFile, extract: Any, dependents: Any = None) -> DependencyGraph: try: data = tomllib.loads(source.text) except tomllib.TOMLDecodeError as error: @@ -284,7 +300,7 @@ def _parse_toml(source: SourceFile, extract: Any) -> DependencyGraph: found: dict[str, Requirement] = {} for requirement in extract(data, source): found[requirement.name] = requirement - return DependencyGraph(found) + return DependencyGraph(found, dependents=None if dependents is None else dependents(data)) def _line_of(source: SourceFile, key: str, default: int = 1) -> int: @@ -335,6 +351,52 @@ def _lock_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requ return found +def _lock_dependents(data: Mapping[str, Any]) -> dict[str, frozenset[str]]: + """For every package a lock file lists or requires, the other locked packages that + require it, optional extras and development groups included. The project's own + packages — uv's editable or virtual sources inside the project — are not others.""" + + dependents: dict[str, set[str]] = {} + for package in data.get("package", []) or []: + name = package.get("name") + if not isinstance(name, str): + continue + dependents.setdefault(normalize(name), set()) + own = _is_project_package(package) + for dependency in _locked_dependencies(package): + found = dependents.setdefault(dependency, set()) + if not own: + found.add(normalize(name)) + return {name: frozenset(found) for name, found in dependents.items()} + + +def _locked_dependencies(package: Mapping[str, Any]) -> Iterator[str]: + declared = package.get("dependencies") or [] + if isinstance(declared, Mapping): # poetry.lock: ``name = specifier``, optional ones too + yield from (normalize(str(name)) for name in declared) + return + groups = ( + declared, + *(package.get("optional-dependencies") or {}).values(), + *(package.get("dev-dependencies") or {}).values(), + ) + for group in groups: + for entry in group or []: + if isinstance(entry, Mapping) and isinstance(entry.get("name"), str): + yield normalize(entry["name"]) + + +def _is_project_package(package: Mapping[str, Any]) -> bool: + source = package.get("source") + if not isinstance(source, Mapping): + return False + path = source.get("editable", source.get("virtual")) + if not isinstance(path, str): + return False + location = PurePosixPath(path) + return not location.is_absolute() and ".." not in location.parts + + class DependencyAnalysis(Analysis[DependencyGraph]): """The project's dependency graph, provided by the engine; empty for a lone file.""" diff --git a/src/coretrace_python/dependency/policy.py b/src/coretrace_python/dependency/policy.py index 2b49aa7..dbc3456 100644 --- a/src/coretrace_python/dependency/policy.py +++ b/src/coretrace_python/dependency/policy.py @@ -29,6 +29,11 @@ class Policy: def denies(self, package: str) -> bool: return normalize(package) in {normalize(name) for name in self.deny} + def accepts(self, finding: Finding) -> bool: + """Whether ``finding`` is about an advisory the project accepts.""" + + return finding.metadata.get("advisory") in self.ignore + def load_policy(path: Path) -> Policy: try: @@ -61,5 +66,4 @@ def _boolean(value: Any, key: str) -> bool: def apply_policy(policy: Policy, findings: Iterable[Finding]) -> tuple[Finding, ...]: """The findings minus those about an advisory the policy accepts.""" - ignored = set(policy.ignore) - return tuple(f for f in findings if f.metadata.get("advisory") not in ignored) + return tuple(f for f in findings if not policy.accepts(f)) diff --git a/src/coretrace_python/dependency/vex.py b/src/coretrace_python/dependency/vex.py new file mode 100644 index 0000000..1a92734 --- /dev/null +++ b/src/coretrace_python/dependency/vex.py @@ -0,0 +1,182 @@ +"""OpenVEX statements from the evidence of a check (architecture §26). + +One statement per advisory affecting a requirement, with the required package as the +subcomponent of the project: + +- ``affected`` when the project's code reaches the vulnerable code: a reachable or + exploitable finding, whatever the policy or a suppression did with it; +- ``not_affected``, as ``vulnerable_code_not_in_execute_path``, only when the advisory is + curated (it names entry points), no code of the project reaches them, every file and + function was analysed, and a lock file shows that no other package requires the + vulnerable one, since the code of installed packages is not analysed; +- ``under_investigation`` otherwise, with the reason in its notes. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Sequence +from datetime import UTC, datetime +from pathlib import Path, PurePosixPath +from urllib.parse import quote + +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.coverage import Coverage + +CONTEXT = "https://openvex.dev/ns/v0.2.0" +# OpenVEX's shared namespace for documents without an IRI of their own, and the author +# OpenVEX tools such as govulncheck write when none is known. +NAMESPACE = "https://openvex.dev/docs/public/vex-" +AUTHOR = "Unknown Author" +# Evidence levels that put the vulnerable code in the project's execution path, lowest first. +REACHED = ("reachable", "exploitable") + + +def render_vex( + dependencies: DependencyGraph, + advisories: Iterable[Advisory], + evidence: Iterable[Finding], + coverage: Coverage, + root: Path, + tool_name: str, + tool_version: str, + timestamp: datetime, +) -> 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 + identifier derives from its statements, not from ``timestamp``.""" + + root = root.resolve() + findings = tuple(evidence) + # The product must be an IRI; the check knows the project by its directory only. + product = f"pkg:generic/{quote(root.name, safe='')}" + statements: list[dict[str, object]] = [] + for advisory in sorted(advisories, key=lambda a: (a.id, a.package)): + about = [ + f + for f in findings + if f.metadata.get("advisory") == advisory.id and f.metadata.get("package") == advisory.package + ] + for requirement in dependencies.requirements: + if advisory.affects(requirement): + statements.append( + { + "vulnerability": _vulnerability(advisory), + "products": [{"@id": product, "subcomponents": [{"@id": purl(requirement)}]}], + **_status(advisory, requirement, about, dependencies, coverage, root), + } + ) + tooling = f"{tool_name} {tool_version}" + content = { + "@context": CONTEXT, + "author": AUTHOR, + "version": 1, + "tooling": tooling, + "statements": statements, + } + digest = hashlib.sha256(json.dumps(content, sort_keys=True).encode("utf-8")).hexdigest() + document = { + "@context": CONTEXT, + "@id": NAMESPACE + digest, + "author": AUTHOR, + "timestamp": timestamp.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"), + "version": 1, + "tooling": tooling, + "statements": statements, + } + return json.dumps(document, indent=2) + "\n" + + +def _vulnerability(advisory: Advisory) -> dict[str, object]: + vulnerability: dict[str, object] = {"name": advisory.id, "description": advisory.summary} + if advisory.aliases: + vulnerability["aliases"] = list(advisory.aliases) + return vulnerability + + +def _status( + advisory: Advisory, + requirement: Requirement, + findings: Sequence[Finding], + dependencies: DependencyGraph, + coverage: Coverage, + root: Path, +) -> dict[str, str]: + package = requirement.name + reached = _reached(findings, root) + if reached: + return { + "status": "affected", + "action_statement": ( + f"Update {package} to a version outside the vulnerable range {advisory.vulnerable}." + ), + "status_notes": f"Reached by the project's code at {reached}.", + } + if not advisory.entry_points: + return _investigating( + "The advisory names no entry point, so whether the project reaches the vulnerable " + "code is not known." + ) + symbols = ", ".join(sorted(str(e.symbol) for e in advisory.entry_points)) + entries = f"the entry points of {advisory.id} ({symbols})" + ruled_out = "; ".join(str(f.metadata["ruled_out"]) for f in findings if "ruled_out" in f.metadata) + ruled = f" Calls ruled out by their arguments: {ruled_out}." if ruled_out else "" + partial = sorted( + _relative(d.path, root) + for d in coverage.details + if d.status != "analysed" or d.analysed < d.functions + ) + if partial: + return _investigating( + f"No analysed code reaches {entries}, but {', '.join(partial)} could not be fully " + f"analysed.{ruled}" + ) + required_by = dependencies.required_by(package) + if required_by is None: + return _investigating( + f"No code of the project reaches {entries}, but no lock file shows which installed " + f"packages require {package}, and their code is not analysed.{ruled}" + ) + if required_by: + return _investigating( + f"No code of the project reaches {entries}, but {package} is required by " + f"{', '.join(sorted(required_by))}, whose code is not analysed.{ruled}" + ) + return { + "status": "not_affected", + "justification": "vulnerable_code_not_in_execute_path", + "impact_statement": ( + f"No code of the project reaches {entries}, and no other locked package requires " + f"{package}.{ruled}" + ), + } + + +def _reached(findings: Sequence[Finding], root: Path) -> str: + """Where the project's code reaches the vulnerable code, each place at its highest + level of evidence; empty when nothing reaches it.""" + + levels: dict[tuple[str, int, str], str] = {} + for finding in findings: + level = finding.metadata.get("level") + if level in REACHED: + path = _relative(str(finding.span.source_id), root) + place = (path, finding.span.start_line, str(finding.metadata.get("symbol"))) + levels[place] = max(level, levels.get(place, level), key=REACHED.index) + return "; ".join( + f"{path}:{line} {symbol} ({level})" for (path, line, symbol), level in sorted(levels.items()) + ) + + +def _investigating(notes: str) -> dict[str, str]: + return {"status": "under_investigation", "status_notes": notes} + + +def _relative(path: str, root: Path) -> str: + try: + return PurePosixPath(Path(path).relative_to(root)).as_posix() + except ValueError: + return path diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index b574597..bd4f8ad 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -201,6 +201,8 @@ class ProjectAnalysis: coverage: Coverage = field(default_factory=Coverage) # Findings silenced by an inline ``# coretrace: ignore`` comment. suppressed: tuple[Finding, ...] = () + # Findings about an advisory the policy accepts, still evidence of what is reached. + accepted: tuple[Finding, ...] = () @dataclass(frozen=True) @@ -474,6 +476,7 @@ def analyze_project( for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): findings = list(apply_refinement(plugin, findings, plugin.refine(context, tuple(findings)))) + accepted = tuple(f for f in findings if policy.accepts(f)) kept, suppressed = partition(apply_policy(policy, findings), _text_of(sources)) return ProjectAnalysis( graph, @@ -485,6 +488,7 @@ def analyze_project( advisories, Coverage(tuple(sorted(coverage, key=lambda c: c.path))), suppressed, + accepted, )