From e6f44e40b693b4560828631ef2ef7a0bd74b9d02 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:10:21 +0200 Subject: [PATCH 1/2] test: define interprocedural taint behavior --- tests/test_interprocedural.py | 12 ++ tests/test_interprocedural_taint.py | 202 ++++++++++++++++++++++++++++ tests/test_taint.py | 15 +-- 3 files changed, 217 insertions(+), 12 deletions(-) create mode 100644 tests/test_interprocedural_taint.py diff --git a/tests/test_interprocedural.py b/tests/test_interprocedural.py index dbbf819..3fd2d5a 100644 --- a/tests/test_interprocedural.py +++ b/tests/test_interprocedural.py @@ -196,6 +196,18 @@ def test_external_calls_propagate_transitively_through_known_callees() -> None: assert reached.call_site is not None and reached.call_site.start_line == 7 +def test_return_values_record_the_external_results_they_depend_on() -> None: + table = summaries( + "def read():\n return input()\n\n" + "def wrap():\n return 'x' + read()\n\n" + "def const():\n return 1\n" + ) + + assert table.summary("read").return_externals == frozenset({SymbolId("python.builtins.input")}) + assert table.summary("wrap").return_externals == frozenset({SymbolId("python.builtins.input")}) + assert table.summary("const").return_externals == frozenset() + + def test_keyword_arguments_are_conservatively_merged() -> None: table = summaries("import subprocess\n\ndef run(cmd, flag):\n subprocess.run(cmd, check=flag)\n") (call,) = table.summary("run").external_calls diff --git a/tests/test_interprocedural_taint.py b/tests/test_interprocedural_taint.py new file mode 100644 index 0000000..944c20f --- /dev/null +++ b/tests/test_interprocedural_taint.py @@ -0,0 +1,202 @@ +"""Acceptance tests for interprocedural taint through function summaries (§17, §19). + +The shared taint engine consumes the call graph and the summaries: a tainted argument +passed to a known function whose parameter reaches a sink is a flow at the call site, +recorded with the function it went through and the sink's own location; the result of a +known function is tainted by the arguments its return value depends on; sanitizing +before the call still works. Nothing is inlined. + +Expected to remain red until ``TaintAnalysis`` requires the interprocedural analyses. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + Sanitizer, + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintFacts, + TaintKind, +) + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +MODELS = ( + Source(SymbolId("python.builtins.input"), "stdin"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + Sink(SymbolId("python.subprocess.run"), TaintKind.COMMAND), + Sanitizer(SymbolId("python.shlex.quote"), TaintKind.COMMAND), +) + + +@pytest.fixture(autouse=True) +def require_interprocedural_taint() -> None: + if not {CallGraphAnalysis, SummaryAnalysis} <= TaintAnalysis.requires: + pytest.fail("TaintAnalysis does not consume the call graph and summaries yet") + + +def facts(source_text: str, name: str = "run") -> TaintFacts: + module = build_hir(SourceManager().add_source("ip.py", source_text)) + manager = AnalysisManager(module) + manager.register(*engine.ALL_ANALYSES) + registry = SecurityModelRegistry() + registry.register(*MODELS) + 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) + + +def lines(result: TaintFacts) -> list[int]: + return [flow.location.start_line for flow in result.flows] + + +# --------------------------------------------------------------------------- parameter sinks + + +def test_tainted_argument_reaching_a_sink_in_a_callee_is_a_flow_at_the_call_site() -> None: + result = facts( + "import os\n\n" + "def execute(command):\n" + " os.system(command)\n\n" + "def run():\n" + " execute(input())\n" + ) + + (flow,) = result.flows + assert flow.location.start_line == 7 + assert flow.sink.symbol == SymbolId("python.os.system") + assert flow.kinds == TaintKind.COMMAND + assert flow.through == "execute" + assert flow.sink_location is not None and flow.sink_location.start_line == 4 + assert flow.source.label == "stdin" + + +def test_direct_flows_have_no_through_function() -> None: + result = facts("import os\n\ndef run():\n os.system(input())\n") + (flow,) = result.flows + assert flow.through is None + assert flow.sink_location == flow.location + + +def test_flows_cross_several_known_calls() -> None: + result = facts( + "import os\n\n" + "def leaf(c):\n os.system(c)\n\n" + "def middle(c):\n leaf(c)\n\n" + "def run():\n middle(input())\n" + ) + + (flow,) = result.flows + assert flow.location.start_line == 10 + assert flow.through == "middle" + assert flow.sink_location is not None and flow.sink_location.start_line == 4 + + +def test_untainted_arguments_to_a_dangerous_callee_are_fine() -> None: + result = facts("import os\n\ndef execute(command):\n os.system(command)\n\ndef run():\n execute('ls')\n") + assert result.flows == () + + +def test_only_the_parameter_that_reaches_the_sink_matters() -> None: + result = facts( + "import os\n\n" + "def execute(command, label):\n print(label)\n os.system(command)\n\n" + "def run():\n execute('ls', input())\n execute(input(), 'x')\n" + ) + assert lines(result) == [9] + + +def test_sanitizing_before_the_call_removes_the_flow() -> None: + result = facts( + "import os\nimport shlex\n\n" + "def execute(command):\n os.system(command)\n\n" + "def run():\n execute(shlex.quote(input()))\n", + ) + assert result.flows == () + + +def test_keyword_arguments_flow_through_summaries() -> None: + result = facts( + "import subprocess\n\n" + "def execute(cmd, check):\n subprocess.run(cmd, check=check)\n\n" + "def run():\n execute(input(), check=True)\n execute('ls', check=input())\n", + ) + assert lines(result) == [7, 8] + + +# --------------------------------------------------------------------------- return taints + + +def test_results_of_known_functions_carry_their_arguments_taint() -> None: + result = facts( + "import os\n\n" + "def wrap(x):\n return 'echo ' + x\n\n" + "def run():\n os.system(wrap(input()))\n" + ) + (flow,) = result.flows + assert flow.location.start_line == 7 + assert flow.through is None + + +def test_results_independent_of_their_arguments_are_clean() -> None: + result = facts("import os\n\ndef const(x):\n return 'ls'\n\ndef run():\n os.system(const(input()))\n") + assert result.flows == () + + +def test_sources_inside_callees_taint_their_results() -> None: + result = facts("import os\n\ndef read():\n return input()\n\ndef run():\n os.system(read())\n") + assert lines(result) == [7] + + +def test_recursive_callees_terminate() -> None: + result = facts( + "import os\n\n" + "def loop(n):\n os.system(n)\n loop(n)\n\n" + "def run():\n loop(input())\n" + ) + assert lines(result) == [8] + + +def test_unsupported_callees_are_conservative() -> None: + result = facts( + "import os\n\n" + "def outer(a):\n def inner():\n pass\n return a\n\n" + "def run():\n os.system(outer(input()))\n" + ) + assert lines(result) == [9] + + +# --------------------------------------------------------------------------- detectors + + +def test_detectors_report_the_function_the_flow_went_through() -> None: + findings = engine.check( + SourceManager().add_source( + "app.py", + "import os\n\ndef execute(command):\n os.system(command)\n\ndef run():\n execute(input())\n", + ), + [PLUGINS], + ) + + (finding,) = findings + assert finding.rule_id == "command-injection" + assert finding.span.start_line == 7 + assert finding.function == "run" + assert finding.message.endswith("reaches python.os.system through execute") + assert finding.metadata["through"] == "execute" + assert finding.metadata["sink_line"] == "4" diff --git a/tests/test_taint.py b/tests/test_taint.py index 42bb19d..d2bf0a3 100644 --- a/tests/test_taint.py +++ b/tests/test_taint.py @@ -18,14 +18,13 @@ import pytest +from coretrace_python import engine from coretrace_python.analysis import AnalysisManager -from coretrace_python.cfg import BlockId, CFGAnalysis, DominanceAnalysis +from coretrace_python.cfg import BlockId, CFGAnalysis from coretrace_python.frontend import build_hir from coretrace_python.hir import nodes -from coretrace_python.ir.lowering import PyIRAnalysis from coretrace_python.ir.model import Call, FunctionIR from coretrace_python.ir.ssa import SSAAnalysis -from coretrace_python.semantic import SEMANTIC_ANALYSES from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import SourceManager @@ -80,15 +79,7 @@ def default_models() -> ModelTable: def analyze(source_text: str, models: ModelTable | None = None) -> tuple[FunctionIR, TaintFacts]: module = build_hir(SourceManager().add_source("taint.py", source_text)) manager = AnalysisManager(module) - manager.register( - *SEMANTIC_ANALYSES, - CFGAnalysis, - DominanceAnalysis, - PyIRAnalysis, - SSAAnalysis, - SecurityModelAnalysis, - TaintAnalysis, - ) + manager.register(*engine.ALL_ANALYSES) manager.provide(SecurityModelAnalysis, default_models() if models is None else models) function = next(s for s in module.body if isinstance(s, nodes.Function)) return manager.get(SSAAnalysis, function), manager.get(TaintAnalysis, function) From 69dd303a125b071865b2bbf75217eef34bfd9f08 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:10:21 +0200 Subject: [PATCH 2/2] feat: make the taint engine consume call graph and summaries --- README.md | 4 +- .../interprocedural/callgraph.py | 4 + .../interprocedural/summaries.py | 79 ++++++---- src/coretrace_python/plugins/detectors.py | 18 ++- src/coretrace_python/taint/engine.py | 135 ++++++++++++++---- 5 files changed, 182 insertions(+), 58 deletions(-) diff --git a/README.md b/README.md index 3245d0e..f9a4e06 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,9 @@ models (`models/python_stdlib`), taint detectors for SQL injection, command inje path traversal, SSRF and XSS (`security/`), and syntactic detectors for `eval`/`exec` and weak hashes (`syntax/`). Model plugins contribute sources, sinks and sanitizers; detectors consume the shared taint result, so adding a framework model makes every -detector aware of it. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing +detector aware of it. Taint follows calls between functions of the same module through +function summaries: a tainted argument passed to a helper that reaches a sink is reported +at the call site, and a helper returning attacker-controlled data taints its result. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing was found, 1 when findings were reported, and 2 on a usage or analysis error. ## Emit PyIR diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 2464859..888ddbd 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -60,6 +60,7 @@ def __init__( ) -> None: self.definitions: Mapping[str, nodes.Function] = MappingProxyType(dict(definitions)) self.functions = tuple(definitions) + self._names = {function.span: name for name, function in definitions.items()} self.unsupported = unsupported self._sites = MappingProxyType(dict(sites)) self._targets = { @@ -72,6 +73,9 @@ def __init__( callers[site.target.name].add(site.caller) self._callers = {name: frozenset(found) for name, found in callers.items()} + def name_of(self, function: nodes.Function) -> str: + return self._names[function.span] + def sites(self, caller: str) -> tuple[CallSite, ...]: return self._sites.get(caller, ()) diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 9358d03..17139cb 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -48,6 +48,20 @@ NONE: Dependencies = frozenset() +@dataclass(frozen=True) +class Dep: + """What a value depends on: parameter indices and results of external symbols.""" + + parameters: Dependencies = NONE + externals: frozenset[SymbolId] = frozenset() + + def __or__(self, other: Dep) -> Dep: + return Dep(self.parameters | other.parameters, self.externals | other.externals) + + +EMPTY = Dep() + + @dataclass(frozen=True) class ExternalCall: """An external symbol reached from this function; dependencies are parameter indices.""" @@ -66,6 +80,7 @@ class FunctionSummary: return_dependencies: Dependencies external_calls: tuple[ExternalCall, ...] unsupported: bool = False + return_externals: frozenset[SymbolId] = frozenset() class SummaryTable: @@ -77,7 +92,7 @@ def summary(self, name: str) -> FunctionSummary: return self._summaries[name] -State = Mapping[Value, Dependencies] +State = Mapping[Value, Dep] _CallKey = tuple[SymbolId, SourceSpan, SourceSpan | None] @@ -93,11 +108,11 @@ def __init__( self.table = table self.blocks = {block.id: block for block in function.blocks} self.external: dict[_CallKey, ExternalCall] = {} - self.returns: Dependencies = NONE + self.returns: Dep = EMPTY def initial(self) -> State: return MappingProxyType( - {p: frozenset({i}) for i, p in enumerate(self.function.parameters)} + {p: Dep(frozenset({i})) for i, p in enumerate(self.function.parameters)} ) def join(self, a: State, b: State) -> State: @@ -108,17 +123,17 @@ def join(self, a: State, b: State) -> State: def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> State: states = list(incoming.values()) - state: dict[Value, Dependencies] = dict(states[0]) if states else {} + state: dict[Value, Dep] = dict(states[0]) if states else {} for other in states[1:]: state = dict(self.join(state, other)) for instruction in block.instructions: if instruction.result is None: continue if isinstance(instruction, Phi): - deps = NONE + deps = EMPTY for value, predecessor in instruction.incoming: if predecessor in incoming: - deps |= incoming[predecessor].get(value, NONE) + deps |= incoming[predecessor].get(value, EMPTY) elif isinstance(instruction, Call): deps = self.call(instruction, state) else: @@ -126,9 +141,9 @@ def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> Stat state[instruction.result] = deps terminator = block.terminator if isinstance(terminator, ForNext) and terminator.result is not None: - state[terminator.result] = state.get(terminator.iterator, NONE) + state[terminator.result] = state.get(terminator.iterator, EMPTY) if isinstance(terminator, Return) and terminator.value is not None: - self.returns |= state.get(terminator.value, NONE) + self.returns |= state.get(terminator.value, EMPTY) return MappingProxyType(state) def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) -> Mapping[BlockId, State]: @@ -146,49 +161,55 @@ def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) - # ------------------------------------------------------------------ transfer @staticmethod - def instruction(instruction: Instruction, state: Mapping[Value, Dependencies]) -> Dependencies: + def instruction(instruction: Instruction, state: Mapping[Value, Dep]) -> Dep: if isinstance(instruction, Constant | Global | Symbol | Undefined): - return NONE - deps = NONE + return EMPTY + deps = EMPTY for operand in instruction.operands(): - deps |= state.get(operand, NONE) + deps |= state.get(operand, EMPTY) return deps - def call(self, call: Call, state: Mapping[Value, Dependencies]) -> Dependencies: - arguments = tuple(state.get(a, NONE) for a in call.arguments) - keywords = NONE + def call(self, call: Call, state: Mapping[Value, Dep]) -> Dep: + arguments = tuple(state.get(a, EMPTY) for a in call.arguments) + keywords = EMPTY for _, value in call.keywords: - keywords |= state.get(value, NONE) + keywords |= state.get(value, EMPTY) everything = keywords for deps in arguments: everything |= deps target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): - self.record(target.symbol, arguments, keywords, call.location, None) - return everything + self.record( + target.symbol, + tuple(a.parameters for a in arguments), + keywords.parameters, + call.location, + None, + ) + return everything | Dep(externals=frozenset({target.symbol})) if isinstance(target, KnownFunction): callee = self.table[target.name] if callee.unsupported: return everything - def mapped(deps: Dependencies) -> Dependencies: - result = NONE - for index in deps: + def mapped(indices: Dependencies) -> Dep: + result = EMPTY + for index in indices: result |= arguments[index] if index < len(arguments) else keywords return result for reached in callee.external_calls: self.record( reached.symbol, - tuple(mapped(d) for d in reached.argument_dependencies), - mapped(reached.keyword_dependencies), + tuple(mapped(d).parameters for d in reached.argument_dependencies), + mapped(reached.keyword_dependencies).parameters, reached.location, call.location, ) - return mapped(callee.return_dependencies) + return mapped(callee.return_dependencies) | Dep(externals=callee.return_externals) assert isinstance(target, UnknownTarget) - return everything | state.get(call.callee, NONE) + return everything | state.get(call.callee, EMPTY) def record( self, @@ -214,12 +235,16 @@ def summarize( problem = _DependenceProblem(name, function, graph, table) solution = solve(problem, cfg) problem.external = {} - problem.returns = NONE + problem.returns = EMPTY for block in function.blocks: if solution.reached(block.id): problem.evaluate(block, solution.incoming(block.id)) return FunctionSummary( - name, len(function.parameters), problem.returns, tuple(problem.external.values()) + name, + len(function.parameters), + problem.returns.parameters, + tuple(problem.external.values()), + return_externals=problem.returns.externals, ) diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index cfc8799..19917f1 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -35,19 +35,25 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: for flow in ctx.get(TaintAnalysis, function).flows: if not flow.kinds & self.kind: continue + message = f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}" + metadata = { + "source": str(flow.source.symbol), + "source_label": flow.source.label, + "sink": str(flow.sink.symbol), + } + if flow.through is not None and flow.sink_location is not None: + message += f" through {flow.through}" + metadata["through"] = flow.through + metadata["sink_line"] = str(flow.sink_location.start_line) findings.append( Finding( rule_id=self.rule_id, - message=f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}", + message=message, severity=self.severity, confidence=Confidence.HIGH, span=flow.location, function=function.name, - metadata={ - "source": str(flow.source.symbol), - "source_label": flow.source.label, - "sink": str(flow.sink.symbol), - }, + metadata=metadata, ) ) return findings diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index de4ac88..74ce95e 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -18,6 +18,14 @@ from coretrace_python.cfg import CFG, BlockId, CFGAnalysis from coretrace_python.dataflow import DataflowProblem, Direction, solve from coretrace_python.hir import nodes +from coretrace_python.interprocedural import ( + CallGraph, + CallGraphAnalysis, + ExternalSymbol, + KnownFunction, + SummaryAnalysis, + SummaryTable, +) from coretrace_python.ir.model import ( BasicBlock, Branch, @@ -65,11 +73,16 @@ def without(self, kinds: TaintKind) -> Taint: @dataclass(frozen=True) class TaintFlow: + """``location`` is the call in the analysed function; when the sink is reached + inside a known callee, ``through`` names it and ``sink_location`` points at the sink.""" + source: Source sink: Sink kinds: TaintKind argument: Value location: SourceSpan + through: str | None = None + sink_location: SourceSpan | None = None class TaintFacts: @@ -87,9 +100,19 @@ def taint(self, value: Value) -> Taint: class _TaintProblem(DataflowProblem[State]): direction: ClassVar[Direction] = Direction.FORWARD - def __init__(self, function: FunctionIR, models: ModelTable) -> None: + def __init__( + self, + name: str, + function: FunctionIR, + models: ModelTable, + graph: CallGraph, + summaries: SummaryTable, + ) -> None: + self.name = name self.function = function self.models = models + self.graph = graph + self.summaries = summaries self.blocks = {block.id: block for block in function.blocks} self.symbols: dict[Value, SymbolId] = { i.result: i.symbol_id @@ -159,30 +182,90 @@ def instruction(self, instruction: Instruction, state: Mapping[Value, Taint]) -> return taint def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) -> Taint: - symbol = self.symbols.get(call.callee) - arguments = Taint.none() - for argument in call.argument_values(): - arguments = arguments.join(state.get(argument, Taint.none())) - if symbol is not None: - sink = self.models.sink(symbol) + arguments = tuple(state.get(a, Taint.none()) for a in call.arguments) + keywords = Taint.none() + for _, value in call.keywords: + keywords = keywords.join(state.get(value, Taint.none())) + everything = keywords + for taint in arguments: + everything = everything.join(taint) + + target = self.graph.target_at(self.name, call.location) + if isinstance(target, ExternalSymbol): + sink = self.models.sink(target.symbol) if sink is not None: for argument in call.argument_values(): - taint = state.get(argument, Taint.none()) - reaching = taint.kinds & sink.kinds - if reaching: - for source in sorted(taint.sources, key=lambda s: str(s.symbol)): - flows.append(TaintFlow(source, sink, reaching, argument, call.location)) - sanitizer = self.models.sanitizer(symbol) + self.report(flows, sink, state.get(argument, Taint.none()), argument, call, None, None) + sanitizer = self.models.sanitizer(target.symbol) if sanitizer is not None: - return arguments.without(sanitizer.kinds) - called_source = self.models.source(symbol) - if called_source is not None: - return arguments.join(Taint(called_source.kinds, frozenset({called_source}))) - return arguments.join(state.get(call.callee, Taint.none())) - - -def propagate_taint(function: FunctionIR, cfg: CFG, models: ModelTable) -> TaintFacts: - problem = _TaintProblem(function, models) + return everything.without(sanitizer.kinds) + source = self.models.source(target.symbol) + if source is not None: + return everything.join(Taint(source.kinds, frozenset({source}))) + return everything + if isinstance(target, KnownFunction): + summary = self.summaries.summary(target.name) + if summary.unsupported: + return everything + + def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: + taint, witness = Taint.none(), None + for index in sorted(deps): + part = arguments[index] if index < len(arguments) else keywords + if part and witness is None: + witness = ( + call.arguments[index] if index < len(arguments) else call.keywords[0][1] + ) + taint = taint.join(part) + return taint, witness + + for reached in summary.external_calls: + sink = self.models.sink(reached.symbol) + if sink is None: + continue + for deps in (*reached.argument_dependencies, reached.keyword_dependencies): + taint, witness = mapped(deps) + if witness is not None: + self.report(flows, sink, taint, witness, call, target.name, reached.location) + result = mapped(summary.return_dependencies)[0] + for symbol in sorted(summary.return_externals, key=str): + returned_source = self.models.source(symbol) + if returned_source is not None: + result = result.join(Taint(returned_source.kinds, frozenset({returned_source}))) + return result + return everything.join(state.get(call.callee, Taint.none())) + + @staticmethod + def report( + flows: list[TaintFlow], + sink: Sink, + taint: Taint, + argument: Value, + call: Call, + through: str | None, + sink_location: SourceSpan | None, + ) -> None: + reaching = taint.kinds & sink.kinds + if not reaching: + return + for source in sorted(taint.sources, key=lambda s: str(s.symbol)): + flows.append( + TaintFlow( + source, + sink, + reaching, + argument, + call.location, + through, + call.location if sink_location is None else sink_location, + ) + ) + + +def propagate_taint( + name: str, function: FunctionIR, cfg: CFG, models: ModelTable, graph: CallGraph, summaries: SummaryTable +) -> TaintFacts: + problem = _TaintProblem(name, function, models, graph, summaries) solution = solve(problem, cfg) taints: dict[Value, Taint] = {} flows: list[TaintFlow] = [] @@ -191,7 +274,7 @@ def propagate_taint(function: FunctionIR, cfg: CFG, models: ModelTable) -> Taint state, found = problem.evaluate(block, solution.incoming(block.id)) taints.update(state) flows.extend(found) - return TaintFacts(taints, tuple(flows)) + return TaintFacts(taints, tuple(dict.fromkeys(flows))) class TaintAnalysis(FunctionAnalysis[TaintFacts]): @@ -199,13 +282,17 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]): name: ClassVar[str] = "taint.flows" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {SSAAnalysis, CFGAnalysis, SecurityModelAnalysis} + {SSAAnalysis, CFGAnalysis, SecurityModelAnalysis, CallGraphAnalysis, SummaryAnalysis} ) @classmethod def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: + graph = ctx.get(CallGraphAnalysis) return propagate_taint( + graph.name_of(function), ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function), ctx.get(SecurityModelAnalysis), + graph, + ctx.get(SummaryAnalysis), )