From 4b6bada3081d96a9f36c9fbd2f5b5d7b1d854f79 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 00:45:06 +0200 Subject: [PATCH 1/2] test: define that a declared sanitizer overrides a project function's summary --- tests/test_interprocedural_taint.py | 15 ++++++++++++-- tests/test_project.py | 32 +++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/tests/test_interprocedural_taint.py b/tests/test_interprocedural_taint.py index 4e1aaca..52e1386 100644 --- a/tests/test_interprocedural_taint.py +++ b/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" diff --git a/tests/test_project.py b/tests/test_project.py index 9b59a59..6ec937b 100644 --- a/tests/test_project.py +++ b/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, From 5a6d90d1f01fe1c172492e60d840aa0b0a947475 Mon Sep 17 00:00:00 2001 From: Hugo Date: Tue, 22 Sep 2026 00:45:06 +0200 Subject: [PATCH 2/2] fix: honour Sanitizer models declared on project functions The taint engine consulted sanitizer models only for external symbols without an interprocedural summary, so a Sanitizer on a function of the analysed project was inert and a code base could not declare its own validation layer. The declared model now takes precedence over the summary derived from the body, for same-file and cross-file calls alike; the flows and side effects the summary reports inside the callee are kept. Closes #95 --- CHANGELOG.md | 6 ++++++ docs/usage.md | 23 +++++++++++++++++++++++ src/coretrace_python/taint/engine.py | 19 ++++++++++++++++--- 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 312ca68..4ea4ab9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### 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 diff --git a/docs/usage.md b/docs/usage.md index 1e60046..6837dfd 100644 --- a/docs/usage.md +++ b/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..`, 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 diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 54dc35d..f3ca52f 100644 --- a/src/coretrace_python/taint/engine.py +++ b/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, )