From 7aa888738d366672ab9f390421dd178b7295c4d7 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 02:32:06 +0200 Subject: [PATCH 1/2] test: specify models a plugin reads from the project under analysis --- tests/test_project_models.py | 117 +++++++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 tests/test_project_models.py diff --git a/tests/test_project_models.py b/tests/test_project_models.py new file mode 100644 index 0000000..0be5740 --- /dev/null +++ b/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 From 1cb57ff668c1de9b2a74b97b5205d2493f4b709c Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 25 Sep 2026 02:32:06 +0200 Subject: [PATCH 2/2] feat: let plugins read models from the project under analysis --- CHANGELOG.md | 1 + docs/plugins.md | 17 +++++++++++++++++ src/coretrace_python/cli.py | 2 ++ src/coretrace_python/engine.py | 10 +++++++--- src/coretrace_python/plugins/api.py | 8 ++++++++ 5 files changed, 35 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00b6122..cf3af74 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/docs/plugins.md b/docs/plugins.md index 9124b91..ec22e4f 100644 --- a/docs/plugins.md +++ b/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 diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index bac85df..776edea 100644 --- a/src/coretrace_python/cli.py +++ b/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, ) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 131f25f..38e5833 100644 --- a/src/coretrace_python/engine.py +++ b/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(): diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 6f052fa..063300f 100644 --- a/src/coretrace_python/plugins/api.py +++ b/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)."""