feat: correlation engine

README.md

@@ -38,7 +38,9 @@ tooling directories (`.venv`, `node_modules`, `__pycache__`, `build`, `dist`) ar
 The dependency files at the root (`requirements*.txt`, `pyproject.toml`, `poetry.lock`,
 `uv.lock`) are resolved into a dependency graph. Plugins may contribute advisories; the
 shipped `dependency/` plugins report a requirement that allows a vulnerable version at
-its line, and every call in the project to an API the advisory affects. The shipped
+its line, and every call in the project to an API the advisory affects. When
+attacker-controlled data reaches such a call, the engine correlates the four facts into
+one critical `exploitable-vulnerability` finding, judged like any other flow. The shipped
 advisory database is a small offline sample; a live OSV feed is future work.
 
 `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be

src/coretrace_python/dependency/correlation.py

@@ -0,0 +1,87 @@
+"""Correlation engine (architecture §27).
+
+A package required in a vulnerable version, an API the advisory affects, a call to that
+API reachable in the project, and attacker-controlled data reaching that call: the
+affected APIs become sinks of the ``ADVISORY`` taint kind, so the shared taint engine,
+the function summaries and the refutation engine do the work, and the flows they leave
+are correlated here into one high-confidence ``exploitable-vulnerability`` finding.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+
+from coretrace_python.dependency.graph import 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
+from coretrace_python.taint import Sink, TaintFlow, TaintKind
+
+
+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] = {}
+    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
+
+
+def advisory_sinks(affected: Mapping[SymbolId, Advisory]) -> tuple[Sink, ...]:
+    return tuple(Sink(symbol, TaintKind.ADVISORY) for symbol in affected)
+
+
+def correlate(
+    function: str,
+    flows: Iterable[TaintFlow],
+    verdicts: Verdicts | None,
+    affected: Mapping[SymbolId, Advisory],
+) -> tuple[Finding, ...]:
+    """Exploitable-vulnerability findings for the non-refuted ADVISORY flows of a function."""
+
+    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,
+            )
+        )
+    return tuple(findings)

src/coretrace_python/engine.py

@@ -26,6 +26,7 @@
     DependencyGraph,
     parse_dependencies,
 )
+from coretrace_python.dependency.correlation import advisory_sinks, affected_symbols, correlate
 from coretrace_python.findings import Confidence, Finding, Severity
 from coretrace_python.findings.refutation import RefutationAnalysis
 from coretrace_python.frontend import HIRBuildError, ParseError, build_hir
@@ -155,7 +156,7 @@ def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ..
     registry = load_plugins(plugin_roots, manager)
     manager.provide(SecurityModelAnalysis, plugin_models(loaded.plugin for loaded in registry))
     manager.provide(ProjectSummaries, SummaryIndex())
-    return _check_module(manager, tuple(loaded.plugin for loaded in registry))
+    return _check_module(manager, tuple(loaded.plugin for loaded in registry))[0]
 
 
 def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph:
@@ -195,8 +196,9 @@ def analyze_project(
     probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("<empty>", "")))
     registry = load_plugins(plugin_roots, probe)
     all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *plugins)
-    models = plugin_models(all_plugins)
     advisories = tuple(a for plugin in all_plugins for a in plugin.advisories)
+    affected = affected_symbols(dependencies, advisories)
+    models = plugin_models(all_plugins).extended(*advisory_sinks(affected))
     index = SummaryIndex()
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
@@ -233,7 +235,20 @@ def analyze_project(
 
     module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin))
     for name in sorted(analysable):
-        findings.extend(_check_module(analysable[name], module_plugins))
+        manager = analysable[name]
+        module_findings, supported = _check_module(manager, module_plugins)
+        findings.extend(module_findings)
+        if affected:
+            graph_of_module = manager.get(CallGraphAnalysis)
+            for function in supported:
+                findings.extend(
+                    correlate(
+                        graph_of_module.name_of(function),
+                        manager.get(TaintAnalysis, function).flows,
+                        manager.get(RefutationAnalysis, function),
+                        affected,
+                    )
+                )
     context = ProjectContext(graph, dependencies, advisories, analysable)
     for plugin in all_plugins:
         if isinstance(plugin, ProjectPlugin):
@@ -246,7 +261,12 @@ def _summaries_of(manager: AnalysisManager) -> dict[str, FunctionSummary]:
     return {name: table.summary(name) for name in table.names}
 
 
-def _check_module(manager: AnalysisManager, plugins: tuple[Plugin, ...]) -> tuple[Finding, ...]:
+def _check_module(
+    manager: AnalysisManager, plugins: tuple[Plugin, ...]
+) -> tuple[tuple[Finding, ...], tuple[nodes.Function, ...]]:
+    """Findings of the module plugins plus notes for unsupported functions, and the
+    functions that could be analysed."""
+
     supported: list[nodes.Function] = []
     notes: list[Finding] = []
     for function in analyzable_functions(manager.module):
@@ -266,7 +286,7 @@ def _check_module(manager: AnalysisManager, plugins: tuple[Plugin, ...]) -> tupl
         else:
             supported.append(function)
     findings = run_plugins(manager, plugins, tuple(supported))
-    return (*findings, *notes)
+    return (*findings, *notes), tuple(supported)
 
 
 def _note(rule_id: str, message: str, source: SourceFile, line: int) -> Finding:

src/coretrace_python/taint/models.py

@@ -25,7 +25,8 @@ class TaintKind(Flag):
     PATH = auto()
     SSRF = auto()
     CODE = auto()
-    ALL = SQL | COMMAND | HTML | PATH | SSRF | CODE
+    ADVISORY = auto()
+    ALL = SQL | COMMAND | HTML | PATH | SSRF | CODE | ADVISORY
 
 
 class ModelError(Exception):
@@ -111,6 +112,17 @@ def sink(self, symbol: SymbolId) -> Sink | None:
         found = self._by_symbol[Sink].get(symbol)
         return found if isinstance(found, Sink) else None
 
+    def extended(self, *sinks: Sink) -> ModelTable:
+        """A table with extra sinks; a sink already present gains the new kinds."""
+
+        merged = {sink.symbol: sink for sink in self.sinks}
+        for sink in sinks:
+            current = merged.get(sink.symbol)
+            merged[sink.symbol] = (
+                Sink(sink.symbol, current.kinds | sink.kinds) if current is not None else sink
+            )
+        return ModelTable(self.sources, tuple(merged.values()), self.sanitizers, self.entry_points)
+
     def sanitizer(self, symbol: SymbolId) -> Sanitizer | None:
         found = self._by_symbol[Sanitizer].get(symbol)
         return found if isinstance(found, Sanitizer) else None