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()) 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: