From 7cef75ac9408d875eefd71174c55adaeb243f2ab Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 00:35:43 +0200 Subject: [PATCH 1/2] test: define how duplicate security models combine across plugins --- tests/test_taint.py | 40 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/tests/test_taint.py b/tests/test_taint.py index d2bf0a3..00e4eac 100644 --- a/tests/test_taint.py +++ b/tests/test_taint.py @@ -131,10 +131,44 @@ def test_registry_indexes_models_by_symbol() -> None: def test_registry_rejects_conflicting_models() -> None: registry = SecurityModelRegistry() - registry.register(Sink(sym("python.os.system"), TaintKind.COMMAND)) + registry.register(Source(sym("python.flask.request.args"), "http")) - with pytest.raises(ModelError, match="python.os.system"): - registry.register(Sink(sym("python.os.system"), TaintKind.SQL)) + with pytest.raises(ModelError, match="python.flask.request.args"): + registry.register(Source(sym("python.flask.request.args"), "cli")) + + +def test_registry_ignores_an_identical_model() -> None: + registry = SecurityModelRegistry() + registry.register(Sanitizer(sym("python.html.escape"), TaintKind.HTML)) + registry.register(Sanitizer(sym("python.html.escape"), TaintKind.HTML)) + + assert registry.freeze().sanitizers == (Sanitizer(sym("python.html.escape"), TaintKind.HTML),) + + +def test_registry_merges_sinks_on_one_symbol() -> None: + registry = SecurityModelRegistry() + registry.register(Sink(sym("python.os.system"), TaintKind.COMMAND, ((TaintKind.COMMAND, (0,)),))) + registry.register(Sink(sym("python.os.system"), TaintKind.SQL)) + + sink = registry.freeze().sink(sym("python.os.system")) + assert sink == Sink( + sym("python.os.system"), TaintKind.COMMAND | TaintKind.SQL, ((TaintKind.COMMAND, (0,)),) + ) + + +def test_plugin_models_names_both_plugins_of_a_conflict() -> None: + from coretrace_python.plugins import ModelPlugin + + class First(ModelPlugin): + name = "first-models" + models = (Source(sym("python.flask.request.args"), "http"),) + + class Second(ModelPlugin): + name = "second-models" + models = (Source(sym("python.flask.request.args"), "cli"),) + + with pytest.raises(ModelError, match="'first-models' and 'second-models'"): + engine.plugin_models([First(), Second()]) def test_models_and_tables_are_immutable() -> None: From b81579b48ecd02904455dc0523f19d687177f5c4 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 00:35:43 +0200 Subject: [PATCH 2/2] fix: merge duplicate security models and name both plugins of a conflict A second model for a symbol aborted the whole run with ModelError, so two plugins could not describe the same symbol. An identical model is now registered once, sinks merge their kinds and positions as ModelTable.extended already did, and a real conflict names both plugins. Closes #92 --- CHANGELOG.md | 6 +++++ docs/plugins.md | 5 ++++ src/coretrace_python/engine.py | 2 +- src/coretrace_python/taint/models.py | 35 +++++++++++++++++++++------- 4 files changed, 38 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 312ca68..ff4fe77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Plugins + +- Two plugins may declare the same symbol: an identical model is registered once and sinks merge their kinds and positions; a real conflict names both plugins instead of failing the run on the first duplicate (#92). + ## 0.2.0 (2026-09-06) ### Adoption in an existing repository diff --git a/docs/plugins.md b/docs/plugins.md index 2013e3b..02f2488 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -142,6 +142,11 @@ class BottleModels(ModelPlugin): | `Validator(symbol, kinds=ALL, argument=0)` | A callable whose truth proves its argument safe, such as `re.fullmatch`; a flow guarded by it is refuted. | | `AuthorizationGuard(symbol, label)` | A decorator or a condition restricting who reaches the code, such as `login_required`; a flow behind it is a hotspot. | +Two plugins may describe the same symbol. An identical model is registered once; two +`Sink` models merge their kinds and positions, so a plugin can add `SQL` to a sink +another plugin declared for `COMMAND`; any other difference is a conflict that stops +the analysis with a `ModelError` naming both plugins. + 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/engine.py b/src/coretrace_python/engine.py index 1b3b90f..498fa1b 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -257,7 +257,7 @@ def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> Plug def plugin_models(plugins: Iterable[Plugin]) -> ModelTable: models = SecurityModelRegistry() for plugin in plugins: - models.register(*plugin.models) + models.register(*plugin.models, origin=plugin.name) return models.freeze() diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 1c212bd..10572f1 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -69,6 +69,11 @@ def kinds_at(self, position: int | None) -> TaintKind: kinds &= ~restricted return kinds + def merged(self, other: Sink) -> Sink: + """This sink with ``other``'s kinds and positions added.""" + + return Sink(self.symbol, self.kinds | other.kinds, self.positions + other.positions) + @dataclass(frozen=True) class Sanitizer: @@ -251,11 +256,7 @@ def extended(self, *sinks: Sink) -> ModelTable: 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, current.positions + sink.positions) - if current is not None - else sink - ) + merged[sink.symbol] = sink if current is None else current.merged(sink) return ModelTable( self.sources, tuple(merged.values()), @@ -279,15 +280,31 @@ class SecurityModelRegistry: def __init__(self) -> None: self._models: dict[tuple[type[Model], SymbolId], Model] = {} + self._origins: dict[tuple[type[Model], SymbolId], str | None] = {} + + def register(self, *models: Model, origin: str | None = None) -> None: + """Add models; ``origin`` names the plugin so a conflict can name both sides. + + Two plugins may describe the same symbol: an identical model is ignored, sinks + merge their kinds and positions, any other difference is a conflict.""" - def register(self, *models: Model) -> None: for model in models: key = (type(model), model.symbol) - if key in self._models: + current = self._models.get(key) + if current is None: + self._models[key] = model + self._origins[key] = origin + elif current == model: + continue + elif isinstance(model, Sink) and isinstance(current, Sink): + self._models[key] = current.merged(model) + else: + registered = self._origins[key] + sides = f" by {registered!r} and {origin!r}" if registered and origin else "" raise ModelError( - f"{type(model).__name__.lower()} model for {model.symbol} is already registered" + f"{type(model).__name__.lower()} model for {model.symbol}" + f" is registered twice{sides} with different values" ) - self._models[key] = model def freeze(self) -> ModelTable: models = list(self._models.values())