fix: honour Sanitizer models declared on project functions

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - 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).
 
+### Precision
+
+- A `Sanitizer` declared on a function of the analysed project takes precedence over the summary derived from its body, in the same file and across files, so a code base can declare its own validation layer (#95).
+
 ## 0.2.0 (2026-09-06)
 
 ### Adoption in an existing repository

docs/usage.md

@@ -335,6 +335,29 @@ coretrace-python-analyzer --check src/ --plugins ./coretrace-plugins
 coretrace-python-analyzer --check src/ --no-bundled-plugins --plugins ./coretrace-plugins
 ```
 
+### Your own validation layer
+
+A code base that already validates its input has functions the analyzer cannot know are
+safe: a `clean()` that returns its argument looks, from its body, like a function that
+passes taint through. Declare them in a model plugin by their project symbol,
+`python.<module>.<function>`, and the analyzer takes your word over what it derived:
+
+```python
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import Sanitizer, TaintKind, Validator
+
+
+class HouseModels(ModelPlugin):
+    name = "house-models"
+    models = (
+        # the result of app.validation.clean is safe for commands, whatever the body does
+        Sanitizer(SymbolId("python.app.validation.clean"), TaintKind.COMMAND),
+        # a flow behind ``if app.validation.is_slug(value):`` is refuted
+        Validator(SymbolId("python.app.validation.is_slug")),
+    )
+```
+
 ## Continuous integration
 
 The exit status gates the job and the SARIF report feeds code scanning. On GitHub

src/coretrace_python/taint/engine.py

@@ -147,8 +147,10 @@ def __init__(
         project: SummaryIndex | None = None,
         heap: HeapFacts | None = None,
         seeds: Mapping[HeapLocation, Taint] | None = None,
+        module: str = "",
     ) -> None:
         self.name = name
+        self.module = module
         self.function = function
         self.models = models
         self.graph = graph
@@ -294,9 +296,10 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T
                 through = target.symbol.canonical_name.removeprefix("python.")
                 bound = () if project.static else self.receiver(call, project.name)
                 arguments = (*(self.deep(r, state) for r in bound), *arguments)
-                return self.known(
+                result = self.known(
                     project, through, arguments, keywords, everything, call, flows, state, (), bound
                 )
+                return self.declared_clean(target.symbol, result)
             symbol = target.symbol
             if not self.modelled(symbol):
                 # ``get_conn().execute`` derived ``app.database.get_conn.execute``; what
@@ -312,9 +315,10 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T
                 *arguments,
                 *(self.deep(value, state) for value in captured),
             )
-            return self.known(
+            result = self.known(
                 summary, target.name, arguments, keywords, everything, call, flows, state, captured, bound
             )
+            return self.declared_clean(project_symbol(self.module, target.name), result)
         returned = self.returned_symbol(call)
         if returned is not None:
             return self.external(returned, everything, call, state, flows)
@@ -369,6 +373,13 @@ def captured(self, call: Call) -> tuple[Value, ...]:
         made = self.defs.get(call.callee)
         return made.captured if isinstance(made, MakeFunction) else ()
 
+    def declared_clean(self, symbol: SymbolId, result: Taint) -> Taint:
+        """A ``Sanitizer`` declared on a project function is the project's own knowledge
+        of what its result carries; it wins over what the summary derived from the body."""
+
+        sanitizer = self.models.sanitizer(symbol)
+        return result if sanitizer is None else result.without(sanitizer.kinds)
+
     def modelled(self, symbol: SymbolId) -> bool:
         return (
             self.models.sink(symbol) is not None
@@ -730,8 +741,9 @@ def propagate_taint(
     project: SummaryIndex | None = None,
     heap: HeapFacts | None = None,
     seeds: Mapping[HeapLocation, Taint] | None = None,
+    module: str = "",
 ) -> TaintFacts:
-    problem = _TaintProblem(name, function, models, graph, summaries, parameters, project, heap, seeds)
+    problem = _TaintProblem(name, function, models, graph, summaries, parameters, project, heap, seeds, module)
     solution = solve(problem, cfg)
     taints: dict[Key, Taint] = {}
     flows: list[TaintFlow] = []
@@ -794,6 +806,7 @@ def sources_of(member: nodes.Function) -> Mapping[int, Source]:
             ctx.get(ProjectSummaries),
             heap,
             seeds,
+            ctx.module.name,
         )
 
 

tests/test_interprocedural_taint.py

@@ -50,12 +50,12 @@ def require_interprocedural_taint() -> None:
         pytest.fail("TaintAnalysis does not consume the call graph and summaries yet")
 
 
-def facts(source_text: str, name: str = "run") -> TaintFacts:
+def facts(source_text: str, name: str = "run", *extra: Sanitizer) -> TaintFacts:
     module = build_hir(SourceManager().add_source("ip.py", source_text))
     manager = AnalysisManager(module)
     manager.register(*engine.ALL_ANALYSES)
     registry = SecurityModelRegistry()
-    registry.register(*MODELS)
+    registry.register(*MODELS, *extra)
     manager.provide(SecurityModelAnalysis, registry.freeze())
     function = next(s for s in module.body if isinstance(s, nodes.Function) and s.name == name)
     return manager.get(TaintAnalysis, function)
@@ -130,6 +130,17 @@ def test_sanitizing_before_the_call_removes_the_flow() -> None:
     assert result.flows == ()
 
 
+def test_a_sanitizer_declared_on_a_project_function_overrides_its_summary() -> None:
+    source = (
+        "import os\n\n"
+        "def clean(value):\n    return value\n\n"
+        "def run():\n    os.system(clean(input()))\n"
+    )
+
+    assert lines(facts(source)) == [7]
+    assert facts(source, "run", Sanitizer(SymbolId("python.ip.clean"), TaintKind.COMMAND)).flows == ()
+
+
 def test_keyword_arguments_flow_through_summaries() -> None:
     result = facts(
         "import subprocess\n\n"

tests/test_project.py

@@ -136,6 +136,38 @@ def test_taint_flows_into_a_function_defined_in_another_file(tmp_path: Path) ->
     assert findings[0].metadata["sink_line"] == "4"
 
 
+def test_a_sanitizer_declared_on_a_function_of_another_file_refutes_the_flow(tmp_path: Path) -> None:
+    root = project(
+        tmp_path / "src",
+        {
+            "app/__init__.py": "",
+            "app/validation.py": "def clean(value):\n    return value\n",
+            "app/main.py": "import os\nfrom app.validation import clean\n\ndef run():\n    os.system(clean(input()))\n",
+        },
+    )
+    models = project(
+        tmp_path / "models",
+        {
+            "plugin.toml": (
+                'name = "house-models"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\n'
+                'requires = []\nprovides = ["models.house"]\n\n'
+                '[entrypoint]\nmodule = "house"\nclass = "HouseModels"\n'
+            ),
+            "house.py": (
+                "from coretrace_python.plugins import ModelPlugin\n"
+                "from coretrace_python.semantic.symbols import SymbolId\n"
+                "from coretrace_python.taint import Sanitizer, TaintKind\n\n\n"
+                "class HouseModels(ModelPlugin):\n"
+                '    name = "house-models"\n'
+                '    models = (Sanitizer(SymbolId("python.app.validation.clean"), TaintKind.COMMAND),)\n'
+            ),
+        },
+    )
+
+    assert rules(engine.analyze_project(root, [PLUGINS]).findings) == [("main.py", "command-injection", 5)]
+    assert rules(engine.analyze_project(root, [PLUGINS, models]).findings) == []
+
+
 def test_module_attribute_calls_resolve_too(tmp_path: Path) -> None:
     root = project(
         tmp_path,