diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d93b2e..75d4cf1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## 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`. +- 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 - 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..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,6 +7,7 @@ from coretrace_python.analysis import AnyAnalysis from coretrace_python.dependency import DependencyAnalysis +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 @@ -17,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.affected_symbols: - affected.setdefault(symbol, advisory) + affected = affected_symbols(ctx.dependencies, ctx.advisories) if not affected: return () findings: list[Finding] = [] @@ -32,25 +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={ - "advisory": advisory.id, - "package": advisory.package, - "symbol": str(site.target.symbol), - }, + 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/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/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..ad2dbd5 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -11,77 +11,103 @@ 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.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.affected_symbols: - affected.setdefault(symbol, advisory) - return affected + for symbol in advisory.reachable_symbols: + 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) +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], 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 = { - "advisory": advisory.id, - "package": advisory.package, - "symbol": str(flow.sink.symbol), - "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/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: diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index f6fdcf4..e62cbb6 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, @@ -481,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()) @@ -524,7 +530,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 +644,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).""" diff --git a/tests/test_advisory_evidence.py b/tests/test_advisory_evidence.py new file mode 100644 index 0000000..3f7fe14 --- /dev/null +++ b/tests/test_advisory_evidence.py @@ -0,0 +1,218 @@ +"""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" + + +# --------------------------------------------------------------------------- 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", + ] + + +# --------------------------------------------------------------------------- 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",)