feat: let plugins read models from the project under analysis

CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Plugins
 
+- `Plugin.project_models(root)` lets a plugin read models from the project under analysis, such as validators the project declares in a file. It is called once per directory check, in every process, and its models are part of the cache key. A model declared wrongly raises `ModelError`, which the command line now reports as an error (exit status 2) instead of a traceback (#140).
 - A `Validator` model may name a function of the project by its project symbol (`python.hc.accounts.views._allow_redirect`), and the refutation recognises it in its own module too, where a call to it has no imported symbol. A project can thus declare its own validation helpers, and a flow they guard is refuted (#134).
 
 ## 0.7.0 (2026-09-25)

docs/plugins.md

@@ -158,6 +158,23 @@ Two plugins may describe the same symbol. An identical model is registered once;
 another plugin declared for `COMMAND`; any other difference is a conflict that stops
 the analysis with a `ModelError` naming both plugins.
 
+A plugin may also read models from the project it analyses: `project_models(self, root)`
+returns them, from a file at `root` for instance. It is called once per directory check,
+in every process analysing it, before any module. Its models join the plugin's own and
+are part of what cached results depend on. A single-file check reads no project. A
+model declared wrongly raises `ModelError`, which stops the check with its message
+(exit status 2). This is how a project declares validators of its own:
+
+```python
+class DeclaredValidators(ModelPlugin):
+    name: ClassVar[str] = "declared-validators"
+
+    def project_models(self, root: Path) -> Sequence[Model]:
+        path = root / "validators.txt"
+        lines = path.read_text(encoding="utf-8").splitlines() if path.is_file() else []
+        return tuple(Validator(SymbolId(line)) for line in lines if line)
+```
+
 A model plugin may also carry `advisories`, a tuple of `Advisory` values
 (`coretrace_python.dependency`) with the package, the vulnerable range and the affected
 symbols, as the shipped `sample-advisories` plugin does. Requirements matching them are

src/coretrace_python/cli.py

@@ -27,6 +27,7 @@
 from coretrace_python.semantic.imports import ImportResolutionError
 from coretrace_python.semantic.scopes import ScopeError
 from coretrace_python.source import SourceManager
+from coretrace_python.taint import ModelError
 
 EXIT_CLEAN = 0
 EXIT_FINDINGS = 1
@@ -45,6 +46,7 @@
     AnalysisError,
     ManifestError,
     IncompatiblePluginError,
+    ModelError,
 )
 
 

src/coretrace_python/engine.py

@@ -281,10 +281,14 @@ def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> Plug
     return registry
 
 
-def plugin_models(plugins: Iterable[Plugin]) -> ModelTable:
+def plugin_models(plugins: Iterable[Plugin], root: Path | None = None) -> ModelTable:
+    """The models of ``plugins``, with those they read from the project at ``root``."""
+
     models = SecurityModelRegistry()
     for plugin in plugins:
         models.register(*plugin.models, origin=plugin.name)
+        if root is not None:
+            models.register(*plugin.project_models(root), origin=plugin.name)
     return models.freeze()
 
 
@@ -392,7 +396,7 @@ def analyze_project(
         (a for plugin in all_plugins for a in plugin.advisories), file_advisories
     )
     affected = affected_symbols(dependencies, advisories)
-    models = plugin_models(all_plugins).extended(*advisory_sinks(affected))
+    models = plugin_models(all_plugins, root).extended(*advisory_sinks(affected))
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
         manager.provide(DependencyAnalysis, dependencies)
@@ -629,7 +633,7 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
         (a for plugin in all_plugins for a in plugin.advisories), file_advisories
     )
     affected = affected_symbols(dependencies, advisories)
-    models = plugin_models(all_plugins).extended(*advisory_sinks(affected))
+    models = plugin_models(all_plugins, batch.root).extended(*advisory_sinks(affected))
     routes = _decode_routes(batch.routes)
     escaped = frozenset(batch.escaped)
     for manager in managers.values():

src/coretrace_python/plugins/api.py

@@ -51,6 +51,14 @@ class Plugin(ABC):
     def analyze(self, ctx: PluginContext) -> Sequence[Finding]:
         raise NotImplementedError
 
+    def project_models(self, root: Path) -> Sequence[Model]:
+        """Models read from the project under ``root``, such as validators it declares
+        in a file there. Called once per directory check, in every process analysing
+        it, before any module; a single-file check reads no project. A model declared
+        wrongly raises ``ModelError``, which stops the check with its message."""
+
+        return ()
+
 
 class ModelPlugin(Plugin):
     """A plugin that only contributes security models or advisories (§15 providers)."""

tests/test_project_models.py

@@ -0,0 +1,117 @@
+"""A plugin may contribute models read from the project it analyses.
+
+``Plugin.project_models(root)`` is called once per directory check, with the project's
+root, in every process that analyses its modules. The models it returns join the
+plugin's own, so a project can declare its validators in a file (``_allow_redirect``,
+which accepts relative URLs only). They are part of what cached results depend on. A
+single-file check reads no project. A model the project declares wrongly stops the
+check with a message, not a traceback.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from coretrace_python import engine
+from coretrace_python.cache import ProjectCache
+from coretrace_python.cli import main
+from coretrace_python.source import SourceManager
+
+MANIFEST = (
+    'name = "declared-validators"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n'
+    'provides = ["model.declared-validators"]\n\n[entrypoint]\nmodule = "declared"\nclass = "Declared"\n'
+)
+PLUGIN = '''
+from pathlib import Path
+from typing import ClassVar
+
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import ModelError, Validator
+
+
+class Declared(ModelPlugin):
+    name: ClassVar[str] = "declared-validators"
+
+    def project_models(self, root: Path):
+        path = root / "validators.txt"
+        if not path.is_file():
+            return ()
+        names = [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+        if any(not name.startswith("python.") for name in names):
+            raise ModelError(f"{path}: a validator is a symbol such as python.app.views.check")
+        return tuple(Validator(SymbolId(name)) for name in names)
+'''
+VIEWS = (
+    "from django.http import HttpRequest\n"
+    "from django.shortcuts import redirect\n\n"
+    "def _allow_redirect(url):\n"
+    "    return url.startswith('/') and not url.startswith('//')\n\n"
+    "def login(request: HttpRequest):\n"
+    "    target = request.GET['next']\n"
+    "    if target and _allow_redirect(target):\n"
+    "        return redirect(target)\n"
+    "    return redirect('/')\n"
+)
+
+
+def setup(tmp_path: Path, declared: str | None) -> tuple[Path, Path]:
+    plugins = tmp_path / "plugins" / "declared"
+    plugins.mkdir(parents=True)
+    (plugins / "plugin.toml").write_text(MANIFEST, encoding="utf-8")
+    (plugins / "declared.py").write_text(PLUGIN, encoding="utf-8")
+    root = tmp_path / "project"
+    (root / "hc").mkdir(parents=True)
+    (root / "hc" / "__init__.py").write_text("", encoding="utf-8")
+    (root / "hc" / "views.py").write_text(VIEWS, encoding="utf-8")
+    if declared is not None:
+        (root / "validators.txt").write_text(declared, encoding="utf-8")
+    return root, tmp_path / "plugins"
+
+
+def redirects(root: Path, plugins: Path, **options: object) -> list[str]:
+    findings = engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins], **options).findings  # type: ignore[arg-type]
+    return [f.metadata["verdict"] for f in findings if f.rule_id == "open-redirect"]
+
+
+def test_a_validator_the_project_declares_refutes_the_flow_it_guards(tmp_path: Path) -> None:
+    root, plugins = setup(tmp_path, "python.hc.views._allow_redirect\n")
+
+    assert redirects(root, plugins) == []
+    (root / "validators.txt").unlink()
+    assert redirects(root, plugins) == ["hotspot"]
+
+
+def test_modules_analysed_in_other_processes_see_the_project_models(tmp_path: Path) -> None:
+    root, plugins = setup(tmp_path, "python.hc.views._allow_redirect\n")
+    (root / "hc" / "other.py").write_text("X = 1\n", encoding="utf-8")
+
+    assert redirects(root, plugins, jobs=2) == []
+
+
+def test_project_models_are_part_of_what_cached_results_depend_on(tmp_path: Path) -> None:
+    root, plugins = setup(tmp_path, "python.hc.views._allow_redirect\n")
+    cache = ProjectCache(tmp_path / "cache")
+
+    first = redirects(root, plugins, cache=cache)
+    (root / "validators.txt").write_text("python.hc.views.other_check\n", encoding="utf-8")
+    second = engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins], cache=cache)
+
+    assert first == []
+    assert second.reused == ()
+    assert [f.metadata["verdict"] for f in second.findings if f.rule_id == "open-redirect"] == ["hotspot"]
+
+
+def test_a_single_file_check_reads_no_project(tmp_path: Path) -> None:
+    root, plugins = setup(tmp_path, "python.hc.views._allow_redirect\n")
+
+    findings = engine.check(SourceManager().load_file(root / "hc" / "views.py"), [engine.BUNDLED_PLUGINS, plugins])
+
+    assert [f.metadata["verdict"] for f in findings if f.rule_id == "open-redirect"] == ["hotspot"]
+
+
+def test_a_model_the_project_declares_wrongly_stops_the_check_with_a_message(tmp_path: Path, capsys) -> None:  # type: ignore[no-untyped-def]
+    root, plugins = setup(tmp_path, "_allow_redirect\n")
+
+    assert main(["--check", str(root), "--plugins", str(plugins)]) == 2
+    assert "a validator is a symbol" in capsys.readouterr().err