fix: merge duplicate security models instead of failing the run

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

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

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

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

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: