From 15d30ed59929081357b504e07ca8a14170c76762 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:34:35 +0200 Subject: [PATCH 1/6] test: define entry points, conditions and the four evidence levels of an advisory --- tests/test_advisory_evidence.py | 157 ++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 tests/test_advisory_evidence.py diff --git a/tests/test_advisory_evidence.py b/tests/test_advisory_evidence.py new file mode 100644 index 0000000..c162b07 --- /dev/null +++ b/tests/test_advisory_evidence.py @@ -0,0 +1,157 @@ +"""What an advisory says beyond a version range, and what a finding keeps of it. + +An advisory names the internal ``affected_symbols`` the fix changed and the public +``entry_points`` through which a project reaches them, each justified by the commit or +by a call path and carrying the ``conditions`` under which it is reached or exploited. +A condition the engine cannot check yet is kept as pending review, never dropped. +Every dependency finding records the highest level of evidence established for it: +``declared`` (the requirement allows a vulnerable version), ``imported`` (a module of +the package is imported), ``reachable`` (an entry point or affected symbol is called), +``exploitable`` (attacker input reaches it). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from coretrace_python import engine +from coretrace_python.dependency import ( + Advisory, + AdvisoryEntryPoint, + Condition, + dump_advisories, + load_advisories, +) +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "src" / "coretrace_python" / "bundled" + +ADVISORY = Advisory( + "CVE-2099-0001", + "vulnlib", + "<1.1", + "parse runs code from the document in unsafe mode", + Severity.CRITICAL, + affected_symbols=(SymbolId("python.vulnlib._parser.parse_unsafe"),), + entry_points=( + AdvisoryEntryPoint( + SymbolId("python.vulnlib.parse"), + "calls _parser.parse_unsafe when mode is UNSAFE (commit 1234abc)", + ( + Condition("argument", "mode is UNSAFE, the default", argument="mode", values=("python.vulnlib.UNSAFE",)), + Condition("semantic", "the document carries a custom tag"), + ), + ), + ), + modules=("vulnlib",), +) + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.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((ADVISORY,)), encoding="utf-8") + return root + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, str, int]]: + return sorted((Path(str(f.span.source_id)).name, f.rule_id, f.span.start_line) for f in findings) + + +def by_rule(findings: tuple[Finding, ...], rule_id: str) -> Finding: + return next(f for f in findings if f.rule_id == rule_id) + + +# --------------------------------------------------------------------------- schema + + +def test_advisory_files_keep_entry_points_conditions_and_modules(tmp_path: Path) -> None: + path = tmp_path / "advisories.json" + path.write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + + assert load_advisories(path) == (ADVISORY,) + entry = json.loads(path.read_text(encoding="utf-8"))["advisories"][0] + assert entry["modules"] == ["vulnlib"] + assert entry["entry_points"][0]["symbol"] == "python.vulnlib.parse" + assert entry["entry_points"][0]["justification"].startswith("calls _parser.parse_unsafe") + assert entry["entry_points"][0]["conditions"][1] == {"kind": "semantic", "text": "the document carries a custom tag"} + + +def test_modules_default_to_the_package_name() -> None: + assert Advisory("X", "python-jose", "<1", "s", Severity.LOW).modules == ("python_jose",) + + +# --------------------------------------------------------------------------- the four levels + + +def test_a_required_package_nobody_imports_is_declared_only(tmp_path: Path) -> None: + root = project(tmp_path, {"requirements.txt": "vulnlib==1.0\n", "app.py": "import os\n"}) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("requirements.txt", "vulnerable-dependency", 1)] + assert findings[0].metadata["level"] == "declared" + + +def test_an_imported_package_without_a_call_is_imported_not_reachable(tmp_path: Path) -> None: + root = project(tmp_path, {"requirements.txt": "vulnlib==1.0\n", "app.py": "import vulnlib\n\nSAFE = vulnlib.SAFE\n"}) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("requirements.txt", "vulnerable-dependency", 1)] + assert findings[0].metadata["level"] == "imported" + + +def test_a_call_to_an_entry_point_is_reachable_with_its_justification_and_conditions(tmp_path: Path) -> None: + root = project( + tmp_path, + {"requirements.txt": "vulnlib==1.0\n", "app.py": "import vulnlib\n\ndef load(text):\n return vulnlib.parse(text)\n"}, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert rules(findings) == [("app.py", "reachable-vulnerability", 4), ("requirements.txt", "vulnerable-dependency", 1)] + reachable = by_rule(findings, "reachable-vulnerability") + assert reachable.metadata["level"] == "reachable" + assert reachable.metadata["entry_point"] == "python.vulnlib.parse" + assert reachable.metadata["justification"].startswith("calls _parser.parse_unsafe") + assert reachable.metadata["conditions"] == "mode is UNSAFE, the default; the document carries a custom tag" + assert "the document carries a custom tag" in reachable.metadata["conditions_pending_review"] + assert by_rule(findings, "vulnerable-dependency").metadata["level"] == "imported" + + +def test_a_call_to_an_affected_symbol_itself_is_reachable_too(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "requirements.txt": "vulnlib==1.0\n", + "app.py": "from vulnlib._parser import parse_unsafe\n\ndef load(text):\n return parse_unsafe(text)\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + reachable = by_rule(findings, "reachable-vulnerability") + assert reachable.metadata["symbol"] == "python.vulnlib._parser.parse_unsafe" + assert reachable.metadata["justification"] == "affected symbol, changed by the fix" + assert "entry_point" not in reachable.metadata + + +def test_attacker_input_reaching_an_entry_point_is_exploitable(tmp_path: Path) -> None: + root = project( + tmp_path, + {"requirements.txt": "vulnlib==1.0\n", "app.py": "import vulnlib\n\ndef load():\n return vulnlib.parse(input())\n"}, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + exploitable = by_rule(findings, "exploitable-vulnerability") + assert exploitable.metadata["level"] == "exploitable" + assert exploitable.metadata["entry_point"] == "python.vulnlib.parse" + assert exploitable.metadata["conditions_pending_review"] != "" + assert by_rule(findings, "reachable-vulnerability").metadata["level"] == "reachable" From f1de9195fa6536726835781ccce50fda8e9afb0e Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:34:35 +0200 Subject: [PATCH 2/6] feat: justify advisory entry points, keep their conditions and record evidence levels An advisory now distinguishes the affected_symbols a fix changed from the public entry_points through which a project reaches them; each entry point carries the justification tying it to the fix and the conditions under which it is affected. A condition the engine cannot check is kept on the finding as pending review, never dropped. modules names what the package installs, so a required package can be told from an imported one. Every dependency finding records its evidence level: declared, imported, reachable or exploitable. Part of CoreTrace-Entreprise/coretrace-python-analyzer-plugins#2 --- CHANGELOG.md | 4 ++ docs/usage.md | 34 ++++++++++- .../reachable_vulnerability.py | 9 +-- .../vulnerable_dependency.py | 3 + src/coretrace_python/dependency/__init__.py | 4 ++ src/coretrace_python/dependency/advisories.py | 42 ++++++++++++- .../dependency/correlation.py | 28 +++++++-- src/coretrace_python/dependency/graph.py | 59 ++++++++++++++++++- 8 files changed, 167 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d93b2e..1818868 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +### Dependencies + +- Advisories distinguish the `affected_symbols` a fix changed from the public `entry_points` through which they are reached, each justified and carrying `conditions` (checkable argument values, or `semantic` ones kept as pending review); `modules` names what the package installs. Every dependency finding records its evidence `level`: `declared`, `imported`, `reachable` or `exploitable`. + ### Plugins - Every plugin module is registered in `sys.modules` before it runs, so a single-file plugin may define a dataclass under `from __future__ import annotations` (#105). diff --git a/docs/usage.md b/docs/usage.md index 966d0e6..bc3d235 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -254,13 +254,45 @@ affected APIs that feed reachability and correlation: "vulnerable": "<5.4", "summary": "yaml.load can execute arbitrary code from untrusted documents", "severity": "critical", - "affected_symbols": ["python.yaml.load", "python.yaml.full_load"], + "affected_symbols": ["python.yaml.constructor.FullConstructor.construct_python_object_apply"], + "entry_points": [ + { + "symbol": "python.yaml.full_load", + "justification": "FullLoader uses FullConstructor (commit 5080ba5)", + "conditions": [ + {"kind": "semantic", "text": "the document carries a python/object/apply tag"} + ] + }, + { + "symbol": "python.yaml.load", + "justification": "same path when Loader is FullLoader", + "conditions": [ + {"kind": "argument", "text": "Loader is FullLoader", "argument": "Loader", "values": ["python.yaml.FullLoader"]}, + {"kind": "semantic", "text": "the document carries a python/object/apply tag"} + ] + } + ], + "modules": ["yaml"], "aliases": ["GHSA-8q59-q68h-6hv4"] } ] } ``` +`affected_symbols` are the functions the fix changed; `entry_points` the public APIs +through which a project reaches them, each with the justification that ties it to the +fix — the commit, or the call path — and the `conditions` under which it is affected: an +argument value the engine can check, or a `semantic` condition it cannot yet, kept on the +finding as pending review rather than dropped. A call to either kind of symbol is +reachable. `modules` names the top-level modules the package installs, so the analyzer +can tell an imported package from a merely required one; it defaults to the package name. + +Every dependency finding records the highest level of evidence established in its +`level` metadata: `declared` (the requirement allows a vulnerable version), `imported` +(a module of the package is imported somewhere), `reachable` (an entry point or affected +symbol is called) or `exploitable` (attacker input reaches it), with `entry_point`, +`justification`, `conditions` and `conditions_pending_review` on the last two. + A `coretrace-policy.toml` at the root, or the file passed with `--policy`, denies packages, requires pins and lists accepted advisories whose findings are dropped: diff --git a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py index a76f5d9..22ef155 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -7,6 +7,7 @@ from coretrace_python.analysis import AnyAnalysis from coretrace_python.dependency import DependencyAnalysis +from coretrace_python.dependency.correlation import evidence from coretrace_python.findings import Confidence, Finding from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol from coretrace_python.plugins import ProjectContext, ProjectPlugin @@ -21,7 +22,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: for requirement in ctx.dependencies.requirements: for advisory in ctx.advisories: if advisory.affects(requirement): - for symbol in advisory.affected_symbols: + for symbol in advisory.reachable_symbols: affected.setdefault(symbol, advisory) if not affected: return () @@ -46,11 +47,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: confidence=Confidence.HIGH, span=site.location, function=function, - metadata={ - "advisory": advisory.id, - "package": advisory.package, - "symbol": str(site.target.symbol), - }, + metadata=evidence(advisory, site.target.symbol, "reachable"), ) ) return findings diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py index 778645d..8c77036 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -17,11 +17,13 @@ class VulnerableDependencyPlugin(ProjectPlugin): def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: findings: list[Finding] = [] + imported = [s for module in ctx.modules for s in ctx.imports(module).all_symbols()] for requirement in ctx.dependencies.requirements: for advisory in ctx.advisories: if not advisory.affects(requirement): continue pinned = requirement.pinned is not None + level = "imported" if advisory.imported_by(imported) else "declared" findings.append( Finding( rule_id="vulnerable-dependency", @@ -37,6 +39,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: "advisory": advisory.id, "package": advisory.package, "specifier": requirement.specifier, + "level": level, }, ) ) diff --git a/src/coretrace_python/dependency/__init__.py b/src/coretrace_python/dependency/__init__.py index 2f7644d..e66aaf9 100644 --- a/src/coretrace_python/dependency/__init__.py +++ b/src/coretrace_python/dependency/__init__.py @@ -11,6 +11,8 @@ from coretrace_python.dependency.graph import ( DEPENDENCY_FILES, Advisory, + AdvisoryEntryPoint, + Condition, DependencyAnalysis, DependencyGraph, Requirement, @@ -26,7 +28,9 @@ "DEPENDENCY_FILES", "POLICY_FILE", "Advisory", + "AdvisoryEntryPoint", "AdvisoryFileError", + "Condition", "DependencyAnalysis", "DependencyGraph", "Policy", diff --git a/src/coretrace_python/dependency/advisories.py b/src/coretrace_python/dependency/advisories.py index 42193cf..ebd39d9 100644 --- a/src/coretrace_python/dependency/advisories.py +++ b/src/coretrace_python/dependency/advisories.py @@ -5,8 +5,8 @@ events into a version specifier; ``dump_advisories`` writes them as a small JSON file that a project keeps at its root as ``advisories.json`` or passes with ``--advisories``. OSV records name no affected APIs, so imported advisories feed the requirement checks -and the SBOM; a file completed by hand with ``affected_symbols`` also feeds the -reachability and correlation checks. +and the SBOM; a file completed by hand with ``affected_symbols``, ``entry_points`` and +their ``conditions`` also feeds the reachability and correlation checks. """ from __future__ import annotations @@ -17,7 +17,7 @@ from pathlib import Path from typing import Any -from coretrace_python.dependency.graph import Advisory, normalize +from coretrace_python.dependency.graph import Advisory, AdvisoryEntryPoint, Condition, normalize from coretrace_python.findings import Severity from coretrace_python.semantic.symbols import SymbolId @@ -137,6 +137,15 @@ def dump_advisories(advisories: Iterable[Advisory]) -> str: "severity": a.severity.value, "affected_symbols": [str(s) for s in a.affected_symbols], "aliases": list(a.aliases), + "entry_points": [ + { + "symbol": str(e.symbol), + "justification": e.justification, + "conditions": [_condition_entry(c) for c in e.conditions], + } + for e in a.entry_points + ], + "modules": list(a.modules), } for a in advisories ], @@ -144,6 +153,15 @@ def dump_advisories(advisories: Iterable[Advisory]) -> str: return json.dumps(document, indent=2) + "\n" +def _condition_entry(condition: Condition) -> dict[str, Any]: + entry: dict[str, Any] = {"kind": condition.kind, "text": condition.text} + if condition.argument is not None: + entry["argument"] = condition.argument + if condition.values: + entry["values"] = list(condition.values) + return entry + + def load_advisories(path: Path) -> tuple[Advisory, ...]: try: document = json.loads(path.read_text(encoding="utf-8")) @@ -165,4 +183,22 @@ def _advisory(entry: Mapping[str, Any]) -> Advisory: Severity(entry["severity"]), tuple(SymbolId(str(s)) for s in entry.get("affected_symbols") or []), tuple(str(a) for a in entry.get("aliases") or []), + tuple(_entry_point(e) for e in entry.get("entry_points") or []), + tuple(str(m) for m in entry.get("modules") or []), + ) + + +def _entry_point(entry: Mapping[str, Any]) -> AdvisoryEntryPoint: + return AdvisoryEntryPoint( + SymbolId(str(entry["symbol"])), + str(entry["justification"]), + tuple( + Condition( + str(c["kind"]), + str(c["text"]), + None if c.get("argument") is None else str(c["argument"]), + tuple(str(v) for v in c.get("values") or []), + ) + for c in entry.get("conditions") or [] + ), ) diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index f700418..fba5344 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -11,7 +11,7 @@ from collections.abc import Iterable, Mapping -from coretrace_python.dependency.graph import Advisory, DependencyGraph +from coretrace_python.dependency.graph import DIRECT, Advisory, DependencyGraph from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.findings.refutation import Status, Verdicts from coretrace_python.semantic.symbols import SymbolId @@ -27,7 +27,7 @@ def affected_symbols( for requirement in dependencies.requirements: for advisory in advisories: if advisory.affects(requirement): - for symbol in advisory.affected_symbols: + for symbol in advisory.reachable_symbols: affected.setdefault(symbol, advisory) return affected @@ -36,6 +36,26 @@ def advisory_sinks(affected: Mapping[SymbolId, Advisory]) -> tuple[Sink, ...]: return tuple(Sink(symbol, TaintKind.ADVISORY) for symbol in affected) +def evidence(advisory: Advisory, symbol: SymbolId, level: str) -> dict[str, str]: + """What a finding keeps of the advisory for ``symbol``: the level of evidence + established, how the symbol relates to the vulnerability and under which + conditions, the ones the engine could not check listed as pending review.""" + + metadata = {"advisory": advisory.id, "package": advisory.package, "symbol": str(symbol), "level": level} + entry = advisory.entry_point(symbol) + if entry is None: + metadata["justification"] = DIRECT + return metadata + metadata["entry_point"] = str(entry.symbol) + metadata["justification"] = entry.justification + if entry.conditions: + metadata["conditions"] = "; ".join(c.text for c in entry.conditions) + # ponytail: no condition is checked yet, so every one awaits review; argument + # conditions get checked at the call site once call sites carry argument symbols. + metadata["conditions_pending_review"] = "; ".join(c.text for c in entry.conditions) + return metadata + + def correlate( function: str, flows: Iterable[TaintFlow], @@ -60,9 +80,7 @@ def correlate( f"the required {advisory.package} {advisory.vulnerable}: {advisory.summary}" ) metadata = { - "advisory": advisory.id, - "package": advisory.package, - "symbol": str(flow.sink.symbol), + **evidence(advisory, flow.sink.symbol, "exploitable"), "source": str(flow.source.symbol), "source_label": flow.source.label, "verdict": "hotspot" if hotspot else "vulnerability", diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 88fde39..4c41f7a 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -11,7 +11,7 @@ import re import tomllib -from collections.abc import Mapping +from collections.abc import Iterable, Mapping from dataclasses import dataclass from pathlib import PurePath from types import MappingProxyType @@ -130,6 +130,37 @@ def may_match(self, vulnerable: str) -> bool: return any(v.satisfies(self.specifier) and v.satisfies(vulnerable) for v in candidates) +@dataclass(frozen=True) +class Condition: + """What must hold for an entry point to reach, or exploit, the affected code: the + value of an argument, the loader or algorithm used, the format decoded, the kind of + input, a configuration. ``kind`` says whether the engine can check it: an + ``argument`` condition names the argument and the values that satisfy it; a + ``semantic`` condition cannot be checked yet and is reported as pending review.""" + + kind: str + text: str + argument: str | None = None + values: tuple[str, ...] = () + + @property + def checkable(self) -> bool: + return self.kind == "argument" + + +@dataclass(frozen=True) +class AdvisoryEntryPoint: + """A public API through which a project reaches an affected symbol, justified by the + fixing commit or by a call path, with the conditions under which it is affected.""" + + symbol: SymbolId + justification: str + conditions: tuple[Condition, ...] = () + + +DIRECT = "affected symbol, changed by the fix" + + @dataclass(frozen=True) class Advisory: id: str @@ -139,10 +170,36 @@ class Advisory: severity: Severity affected_symbols: tuple[SymbolId, ...] = () aliases: tuple[str, ...] = () + entry_points: tuple[AdvisoryEntryPoint, ...] = () + # Top-level modules the package installs, to tell an imported package from a merely + # required one; defaults to the package name (``pyyaml`` installs ``yaml``: say so). + modules: tuple[str, ...] = () + + def __post_init__(self) -> None: + if not self.modules: + object.__setattr__(self, "modules", (normalize(self.package).replace("-", "_"),)) def affects(self, requirement: Requirement) -> bool: return requirement.name == normalize(self.package) and requirement.may_match(self.vulnerable) + @property + def reachable_symbols(self) -> tuple[SymbolId, ...]: + """Every symbol a call to which reaches the vulnerability: the entry points and + the affected symbols themselves.""" + + return (*(e.symbol for e in self.entry_points), *self.affected_symbols) + + def entry_point(self, symbol: SymbolId) -> AdvisoryEntryPoint | None: + return next((e for e in self.entry_points if e.symbol == symbol), None) + + def imported_by(self, symbols: Iterable[SymbolId]) -> bool: + """Whether any of the ``symbols`` a project imports belongs to this package.""" + + prefixes = tuple(f"python.{m}" for m in self.modules) + return any( + s.canonical_name == p or s.canonical_name.startswith(p + ".") for s in symbols for p in prefixes + ) + class DependencyGraph: def __init__(self, requirements: Mapping[str, Requirement] | None = None, errors: tuple[str, ...] = ()) -> None: From 6fe5817fb62c6afc2fe44f1b7f1a6c54a5767b05 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:41:10 +0200 Subject: [PATCH 3/6] test: define that every advisory matching a call is reported --- tests/test_advisory_evidence.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_advisory_evidence.py b/tests/test_advisory_evidence.py index c162b07..f0078e2 100644 --- a/tests/test_advisory_evidence.py +++ b/tests/test_advisory_evidence.py @@ -155,3 +155,34 @@ def test_attacker_input_reaching_an_entry_point_is_exploitable(tmp_path: Path) - assert exploitable.metadata["entry_point"] == "python.vulnlib.parse" assert exploitable.metadata["conditions_pending_review"] != "" assert by_rule(findings, "reachable-vulnerability").metadata["level"] == "reachable" + + +# --------------------------------------------------------------------------- several advisories on one symbol + + +def test_every_advisory_matching_a_call_is_reported(tmp_path: Path) -> None: + second = Advisory( + "CVE-2099-0002", + "vulnlib", + "<1.0.5", + "parse leaks memory on a crafted document", + Severity.HIGH, + entry_points=(AdvisoryEntryPoint(SymbolId("python.vulnlib.parse"), "parse reads the document (commit 9abc)"),), + modules=("vulnlib",), + ) + root = project( + tmp_path, + {"requirements.txt": "vulnlib==1.0\n", "app.py": "import vulnlib\n\ndef load():\n return vulnlib.parse(input())\n"}, + ) + (root / "advisories.json").write_text(dump_advisories((ADVISORY, second)), encoding="utf-8") + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert sorted(f.metadata["advisory"] for f in findings if f.rule_id == "reachable-vulnerability") == [ + "CVE-2099-0001", + "CVE-2099-0002", + ] + assert sorted(f.metadata["advisory"] for f in findings if f.rule_id == "exploitable-vulnerability") == [ + "CVE-2099-0001", + "CVE-2099-0002", + ] From 00daa2f7a4595ecc14acc2f02e113165a6579484 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:41:10 +0200 Subject: [PATCH 4/6] fix: report every advisory an affected symbol matches The affected-symbol map kept the first advisory per symbol, so a call to yaml.load under a release affected by four CVEs reported one. The map now carries every advisory of the symbol; the reachable plugin and the correlation report one finding per advisory. --- CHANGELOG.md | 1 + .../reachable_vulnerability.py | 39 ++++---- .../dependency/correlation.py | 88 ++++++++++--------- src/coretrace_python/engine.py | 11 ++- 4 files changed, 73 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1818868..217d306 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### Dependencies - Advisories distinguish the `affected_symbols` a fix changed from the public `entry_points` through which they are reached, each justified and carrying `conditions` (checkable argument values, or `semantic` ones kept as pending review); `modules` names what the package installs. Every dependency finding records its evidence `level`: `declared`, `imported`, `reachable` or `exploitable`. +- A call to an API affected by several advisories of the pinned release reports every one of them, not the first listed. ### Plugins diff --git a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py index 22ef155..f66a402 100644 --- a/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py +++ b/src/coretrace_python/bundled/dependency/reachable_vulnerability/reachable_vulnerability.py @@ -7,7 +7,7 @@ from coretrace_python.analysis import AnyAnalysis from coretrace_python.dependency import DependencyAnalysis -from coretrace_python.dependency.correlation import evidence +from coretrace_python.dependency.correlation import affected_symbols, evidence from coretrace_python.findings import Confidence, Finding from coretrace_python.interprocedural import CallGraphAnalysis, ExternalSymbol from coretrace_python.plugins import ProjectContext, ProjectPlugin @@ -18,12 +18,7 @@ class ReachableVulnerabilityPlugin(ProjectPlugin): requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({DependencyAnalysis, CallGraphAnalysis}) def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: - affected = {} - for requirement in ctx.dependencies.requirements: - for advisory in ctx.advisories: - if advisory.affects(requirement): - for symbol in advisory.reachable_symbols: - affected.setdefault(symbol, advisory) + affected = affected_symbols(ctx.dependencies, ctx.advisories) if not affected: return () findings: list[Finding] = [] @@ -33,21 +28,19 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: for site in graph.sites(function): if not isinstance(site.target, ExternalSymbol): continue - advisory = affected.get(site.target.symbol) - if advisory is None: - continue - findings.append( - Finding( - rule_id="reachable-vulnerability", - message=( - f"{advisory.id}: {site.target.symbol} is affected in the required " - f"{advisory.package} {advisory.vulnerable}: {advisory.summary}" - ), - severity=advisory.severity, - confidence=Confidence.HIGH, - span=site.location, - function=function, - metadata=evidence(advisory, site.target.symbol, "reachable"), + for advisory in affected.get(site.target.symbol, ()): + findings.append( + Finding( + rule_id="reachable-vulnerability", + message=( + f"{advisory.id}: {site.target.symbol} is affected in the required " + f"{advisory.package} {advisory.vulnerable}: {advisory.summary}" + ), + severity=advisory.severity, + confidence=Confidence.HIGH, + span=site.location, + function=function, + metadata=evidence(advisory, site.target.symbol, "reachable"), + ) ) - ) return findings diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index fba5344..ad2dbd5 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -13,26 +13,30 @@ from coretrace_python.dependency.graph import DIRECT, Advisory, DependencyGraph from coretrace_python.findings import Confidence, Finding, Severity -from coretrace_python.findings.refutation import Status, Verdicts +from coretrace_python.findings.refutation import Status, Verdict, Verdicts from coretrace_python.semantic.symbols import SymbolId from coretrace_python.taint import Sink, TaintFlow, TaintKind +Affected = Mapping[SymbolId, tuple[Advisory, ...]] -def affected_symbols( - dependencies: DependencyGraph, advisories: Iterable[Advisory] -) -> Mapping[SymbolId, Advisory]: - """The APIs affected by advisories whose package is required in a vulnerable version.""" - affected: dict[SymbolId, Advisory] = {} +def affected_symbols(dependencies: DependencyGraph, advisories: Iterable[Advisory]) -> Affected: + """The APIs affected by advisories whose package is required in a vulnerable version, + each with every such advisory: a call to ``yaml.load`` reaches every CVE of the + pinned release, not the first one listed.""" + + affected: dict[SymbolId, list[Advisory]] = {} for requirement in dependencies.requirements: for advisory in advisories: if advisory.affects(requirement): for symbol in advisory.reachable_symbols: - affected.setdefault(symbol, advisory) - return affected + found = affected.setdefault(symbol, []) + if advisory not in found: + found.append(advisory) + return {symbol: tuple(found) for symbol, found in affected.items()} -def advisory_sinks(affected: Mapping[SymbolId, Advisory]) -> tuple[Sink, ...]: +def advisory_sinks(affected: Affected) -> tuple[Sink, ...]: return tuple(Sink(symbol, TaintKind.ADVISORY) for symbol in affected) @@ -60,46 +64,50 @@ def correlate( function: str, flows: Iterable[TaintFlow], verdicts: Verdicts | None, - affected: Mapping[SymbolId, Advisory], + affected: Affected, ) -> tuple[Finding, ...]: - """Exploitable-vulnerability findings for the non-refuted ADVISORY flows of a function.""" + """Exploitable-vulnerability findings for the non-refuted ADVISORY flows of a function, + one per advisory the sink is affected by.""" findings: list[Finding] = [] for flow in flows: if not flow.kinds & TaintKind.ADVISORY: continue - advisory = affected.get(flow.sink.symbol) - if advisory is None: - continue verdict = verdicts.verdict(flow) if verdicts is not None else None if verdict is not None and verdict.status is Status.REFUTED: continue hotspot = verdict is not None and verdict.status is Status.HOTSPOT - message = ( - f"{advisory.id}: {flow.source.label} input reaches {flow.sink.symbol}, affected in " - f"the required {advisory.package} {advisory.vulnerable}: {advisory.summary}" - ) - metadata = { - **evidence(advisory, flow.sink.symbol, "exploitable"), - "source": str(flow.source.symbol), - "source_label": flow.source.label, - "verdict": "hotspot" if hotspot else "vulnerability", - } - if verdict is not None: - metadata["evidence"] = verdict.evidence - if flow.through is not None and flow.sink_location is not None: - message += f" through {flow.through}" - metadata["through"] = flow.through - metadata["sink_line"] = str(flow.sink_location.start_line) - findings.append( - Finding( - rule_id="exploitable-vulnerability", - message=message, - severity=Severity.CRITICAL, - confidence=Confidence.MEDIUM if hotspot else Confidence.HIGH, - span=flow.location, - function=function, - metadata=metadata, - ) + findings.extend( + _exploitable(function, flow, advisory, verdict, hotspot) for advisory in affected.get(flow.sink.symbol, ()) ) return tuple(findings) + + +def _exploitable( + function: str, flow: TaintFlow, advisory: Advisory, verdict: Verdict | None, hotspot: bool +) -> Finding: + message = ( + f"{advisory.id}: {flow.source.label} input reaches {flow.sink.symbol}, affected in " + f"the required {advisory.package} {advisory.vulnerable}: {advisory.summary}" + ) + metadata = { + **evidence(advisory, flow.sink.symbol, "exploitable"), + "source": str(flow.source.symbol), + "source_label": flow.source.label, + "verdict": "hotspot" if hotspot else "vulnerability", + } + if verdict is not None: + metadata["evidence"] = verdict.evidence + if flow.through is not None and flow.sink_location is not None: + message += f" through {flow.through}" + metadata["through"] = flow.through + metadata["sink_line"] = str(flow.sink_location.start_line) + return Finding( + rule_id="exploitable-vulnerability", + message=message, + severity=Severity.CRITICAL, + confidence=Confidence.MEDIUM if hotspot else Confidence.HIGH, + span=flow.location, + function=function, + metadata=metadata, + ) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index f6fdcf4..a54c93d 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -48,7 +48,12 @@ load_policy, parse_dependencies, ) -from coretrace_python.dependency.correlation import advisory_sinks, affected_symbols, correlate +from coretrace_python.dependency.correlation import ( + Affected, + advisory_sinks, + affected_symbols, + correlate, +) from coretrace_python.findings import ( FINDING_SCHEMA_VERSION, Confidence, @@ -524,7 +529,7 @@ def _analyse_managers( managers: Mapping[str, AnalysisManager], seed: SummaryIndex, plugins: tuple[Plugin, ...], - affected: Mapping[SymbolId, Advisory], + affected: Affected, ) -> dict[str, CachedModule]: """Analyse one component: iterate its summaries to a fixpoint over ``seed`` (§21), then extract what the rest of the run needs from each module.""" @@ -638,7 +643,7 @@ def _configuration_key( def _analyse_module( - manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Mapping[SymbolId, Advisory] + manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Affected ) -> CachedModule: """One module's findings, summaries and call sites: what the cache keeps (§11).""" From a78e8b0814ab10ecfb1fa3476beb56a986017c1b Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:44:50 +0200 Subject: [PATCH 5/6] test: define that a later contributor replaces an advisory by identifier --- tests/test_advisory_evidence.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/test_advisory_evidence.py b/tests/test_advisory_evidence.py index f0078e2..3f7fe14 100644 --- a/tests/test_advisory_evidence.py +++ b/tests/test_advisory_evidence.py @@ -186,3 +186,33 @@ def test_every_advisory_matching_a_call_is_reported(tmp_path: Path) -> None: "CVE-2099-0001", "CVE-2099-0002", ] + + +# --------------------------------------------------------------------------- one advisory per identifier + + +def test_a_later_contributor_replaces_an_advisory_by_identifier(tmp_path: Path) -> None: + """The bundled sample says CVE-2099-0001 is ``<2``; the project's curated file says + ``<1.1`` with entry points. One advisory survives, the curated one.""" + + from coretrace_python.plugins import ModelPlugin + + class Sample(ModelPlugin): + name = "sample" + advisories = (Advisory("CVE-2099-0001", "vulnlib", "<2", "coarse", Severity.LOW),) + + root = project(tmp_path, {"requirements.txt": "vulnlib==1.5\n", "app.py": "import vulnlib\n\ndef f(t):\n return vulnlib.parse(t)\n"}) + + findings = engine.analyze_project(root, [PLUGINS], plugins=[Sample()]).findings + + assert [f.metadata["advisory"] for f in findings] == [] # 1.5 is outside the curated range + + +def test_the_sample_advisories_name_the_modules_their_packages_install() -> None: + from coretrace_python.bundled.dependency.sample_advisories.sample_advisories import ( + SampleAdvisories, + ) + + modules = {a.package: a.modules for a in SampleAdvisories.advisories} + assert modules["pyyaml"] == ("yaml",) + assert modules["pillow"] == ("PIL",) From 963059d52436c91f01ebe6518897708ffe324c81 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 16:44:50 +0200 Subject: [PATCH 6/6] fix: merge advisories by identifier and name the sample packages' modules Two contributions of the same advisory with different ranges both survived and produced two findings. The later contributor now replaces the earlier one, so a curated plugin refines the bundled sample and a local file wins over every plugin. The sample advisories say which modules pyyaml and pillow install. --- CHANGELOG.md | 1 + .../dependency/sample_advisories/sample_advisories.py | 2 ++ src/coretrace_python/engine.py | 9 +++++---- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 217d306..75d4cf1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - Advisories distinguish the `affected_symbols` a fix changed from the public `entry_points` through which they are reached, each justified and carrying `conditions` (checkable argument values, or `semantic` ones kept as pending review); `modules` names what the package installs. Every dependency finding records its evidence `level`: `declared`, `imported`, `reachable` or `exploitable`. - A call to an API affected by several advisories of the pinned release reports every one of them, not the first listed. +- One advisory per identifier and package: a later contributor replaces an earlier one, so a curated feed refines the bundled sample, and the sample names the modules its packages install. ### Plugins diff --git a/src/coretrace_python/bundled/dependency/sample_advisories/sample_advisories.py b/src/coretrace_python/bundled/dependency/sample_advisories/sample_advisories.py index 6530cae..3b52e0e 100644 --- a/src/coretrace_python/bundled/dependency/sample_advisories/sample_advisories.py +++ b/src/coretrace_python/bundled/dependency/sample_advisories/sample_advisories.py @@ -28,6 +28,7 @@ class SampleAdvisories(ModelPlugin): "yaml.load and full_load can execute arbitrary code from untrusted documents", Severity.CRITICAL, (_sym("yaml.load"), _sym("yaml.full_load"), _sym("yaml.unsafe_load")), + modules=("yaml",), ), Advisory( "CVE-2018-18074", @@ -105,5 +106,6 @@ class SampleAdvisories(ModelPlugin): Severity.HIGH, (_sym("PIL.Image.open"),), ("GHSA-cfh3-3jmp-rvhc",), + modules=("PIL",), ), ) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index a54c93d..e62cbb6 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -486,12 +486,13 @@ def analyze_project( def _merge_advisories( from_plugins: Iterable[Advisory], from_files: Iterable[Advisory] ) -> tuple[Advisory, ...]: - """One advisory per identifier, package and range; a local file's version wins over - a plugin's, since the file is the project's own curated feed.""" + """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.""" - merged: dict[tuple[str, str, str], Advisory] = {} + merged: dict[tuple[str, str], Advisory] = {} for advisory in (*from_plugins, *from_files): - merged[(advisory.id, advisory.package, advisory.vulnerable)] = advisory + merged[(advisory.id, advisory.package)] = advisory return tuple(merged.values())