diff --git a/README.md b/README.md index b03179a..b2a40f0 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,12 @@ allocation site one abstract object with an `elements` and an `attributes` locat `insert`, `add`, `update` and iteration all carry taint through the container. Function summaries record the parameters a function mutates and the module globals it touches, so a helper that fills a list taints the caller's list, across files too. +Objects of the project's own classes are followed too: `App(x)` calls `App.__init__` on +the new object, `app.run()` and `self.run()` call the class's methods with the receiver as +`self`, so a value stored in one method and read in another flows through the object, +across files through the project index. A method the framework calls, a Django view's +`get` for instance, starts with what its sibling methods store into `self` from their +own inputs. Each flow is then judged: a dominating guard that proves the value safe (`isdigit()`, membership in a constant allowlist, equality with a constant) refutes it and it is not reported; a guard that only mentions the value makes it a hotspot with medium confidence; diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index f8514c0..7573ba9 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -151,29 +151,48 @@ def resolve_targets( known: frozenset[str], parameters: Mapping[Value, SymbolId] | None = None, nested: frozenset[str] = frozenset(), + classes: Mapping[str, frozenset[str]] | None = None, + owner: str | None = None, + typed: Mapping[Value, str] | None = None, ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]: """Map every callee value of ``function`` to its target, and every symbol value. - ``nested`` names the functions defined inside this one, ``outer.inner``.""" + ``nested`` names the functions defined inside this one (``outer.inner``); + ``classes`` maps the module's classes to their methods, ``owner`` is the class of a + method and ``typed`` the parameters annotated with a module class. ``App(x)`` is a + call to ``App.__init__``, ``app.run()`` and ``self.run()`` calls to ``App.run``.""" module = scopes.module_scope symbols = derive_symbols(function, parameters) targets: dict[Value, Target] = { value: ExternalSymbol(symbol) for value, symbol in symbols.items() } + classes = classes or {} + instance_of: dict[Value, str] = dict(typed or {}) + if owner is not None and function.parameters: + instance_of[function.parameters[0]] = owner + defs = {i.result: i for block in function.blocks for i in block.instructions if i.result} for block in function.blocks: for instruction in block.instructions: if isinstance(instruction, MakeFunction): qualified = f"{function.name}.{instruction.name}" if qualified in nested: targets[instruction.result] = KnownFunction(qualified) - if isinstance(instruction, Global): + elif isinstance(instruction, Global): binding = module.bindings.get(instruction.name) - if ( - binding is not None - and binding.kind is BindingKind.FUNCTION - and instruction.name in known - ): + if binding is None: + continue + if binding.kind is BindingKind.FUNCTION and instruction.name in known: targets[instruction.result] = KnownFunction(instruction.name) + elif binding.kind is BindingKind.CLASS and "__init__" in classes.get(instruction.name, ()): + targets[instruction.result] = KnownFunction(f"{instruction.name}.__init__") + elif isinstance(instruction, Call): + callee = defs.get(instruction.callee) + if isinstance(callee, Global) and callee.name in classes: + instance_of[instruction.result] = callee.name + elif isinstance(instruction, GetAttr) and instruction.object in instance_of: + method = f"{instance_of[instruction.object]}.{instruction.attribute}" + if method in nested: + targets[instruction.result] = KnownFunction(method) return targets, symbols @@ -195,6 +214,19 @@ def _annotated( return found +def _typed_with_module_classes( + function: nodes.Function, ssa: FunctionIR, classes: Mapping[str, frozenset[str]] +) -> dict[Value, str]: + """Parameters annotated with a class of this module (``a: App``).""" + + found: dict[Value, str] = {} + for value, parameter in zip(ssa.parameters, function.parameters, strict=False): + annotation = parameter.annotation + if isinstance(annotation, nodes.Name) and annotation.identifier in classes: + found[value] = annotation.identifier + return found + + class CallGraphAnalysis(Analysis[CallGraph]): name: ClassVar[str] = "interprocedural.callgraph" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis, SymbolAnalysis}) @@ -216,6 +248,11 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: known = frozenset( name for name, function in definitions.items() if function in ctx.module.body ) + classes = { + statement.name: frozenset(m.name for m in statement.body if isinstance(m, nodes.Function)) + for statement in ctx.module.body + if isinstance(statement, nodes.Class) + } sites: dict[str, tuple[CallSite, ...]] = {} symbols: dict[str, Mapping[Value, SymbolId]] = {} unsupported: set[str] = set() @@ -226,8 +263,16 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: unsupported.add(name) sites[name] = () continue + owner = name.rsplit(".", 1)[0] if "." in name and name.rsplit(".", 1)[0] in classes else None targets, symbols[name] = resolve_targets( - ssa, scopes, known, _annotated(function, ssa, scopes, table), frozenset(definitions) + ssa, + scopes, + known, + _annotated(function, ssa, scopes, table), + frozenset(definitions), + classes, + owner, + _typed_with_module_classes(function, ssa, classes), ) found: list[CallSite] = [] for block in ssa.blocks: diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index e00b851..f70ef84 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -300,8 +300,12 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: if isinstance(target, ExternalSymbol): project = self.project.get(target.symbol) + if project is None: + project = self.project.get(target.symbol.attribute("__init__")) if project is not None: - return self.known(project, arguments, keywords, everything, call, state) + bound = self.receiver(call, project.name) + arguments = (*(self.deep(r, state) for r in bound), *arguments) + return self.known(project, arguments, keywords, everything, call, state, bound) self.record( target.symbol, tuple(a.parameters for a in arguments), @@ -314,10 +318,25 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: made = self.defs.get(call.callee) if isinstance(made, MakeFunction): arguments = (*arguments, *(self.deep(v, state) for v in made.captured)) - return self.known(self.table[target.name], arguments, keywords, everything, call, state) + bound = self.receiver(call, target.name) + arguments = (*(self.deep(r, state) for r in bound), *arguments) + return self.known(self.table[target.name], arguments, keywords, everything, call, state, bound) assert isinstance(target, UnknownTarget) return everything | state.get(call.callee, EMPTY) + def receiver(self, call: Call, name: str) -> tuple[Value, ...]: + """The object a method call runs on: the receiver of ``obj.method(...)`` or the + new object of ``Class(...)``, the method's implicit first parameter.""" + + if "." not in name: + return () + callee = self.defs.get(call.callee) + if isinstance(callee, GetAttr): + return (callee.object,) + if name.endswith(".__init__") and isinstance(callee, Global | Symbol): + return (call.result,) + return () + def known( self, callee: FunctionSummary, @@ -326,6 +345,7 @@ def known( everything: Dep, call: Call, state: dict[Key, Dep], + receiver: tuple[Value, ...] = (), ) -> Dep: """Dependencies of a call to a function whose summary is known.""" @@ -347,7 +367,7 @@ def mapped(indices: Dependencies) -> Dep: call.location, ) made = self.defs.get(call.callee) - values = (*call.arguments, *(made.captured if isinstance(made, MakeFunction) else ())) + values = (*receiver, *call.arguments, *(made.captured if isinstance(made, MakeFunction) else ())) for mutation in callee.mutations: if mutation.parameter < len(values): deps = mapped(mutation.dependencies) | Dep(externals=mutation.externals) diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 434e528..518ecaa 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -9,7 +9,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from types import MappingProxyType from typing import ClassVar @@ -53,6 +53,7 @@ GetAttr, GetItem, GetIter, + Global, Instruction, Jump, MakeFunction, @@ -144,6 +145,7 @@ def __init__( parameters: Mapping[int, Source] | None = None, project: SummaryIndex | None = None, heap: HeapFacts | None = None, + seeds: Mapping[HeapLocation, Taint] | None = None, ) -> None: self.name = name self.function = function @@ -153,6 +155,7 @@ def __init__( self.parameters = parameters or {} self.project = project or SummaryIndex() self.heap = heap or HeapFacts({}) + self.seeds = dict(seeds or {}) self.blocks = {block.id: block for block in function.blocks} self.defs: dict[Value, Instruction] = { i.result: i for block in function.blocks for i in block.instructions if i.result @@ -187,13 +190,14 @@ def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, T return taint def initial(self) -> State: - return MappingProxyType( - { - self.function.parameters[index]: Taint(source.kinds, frozenset({source})) - for index, source in self.parameters.items() - if index < len(self.function.parameters) - } - ) + state: dict[Key, Taint] = { + self.function.parameters[index]: Taint(source.kinds, frozenset({source})) + for index, source in self.parameters.items() + if index < len(self.function.parameters) + } + for location, taint in self.seeds.items(): + state[location] = taint + return MappingProxyType(state) def join(self, a: State, b: State) -> State: merged = dict(a) @@ -280,9 +284,16 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T target = self.graph.target_at(self.name, call.location) if isinstance(target, ExternalSymbol): project = self.project.summary(target.symbol) + if project is None: + # ``App(x)`` with ``App`` defined in another file: its ``__init__``. + project = self.project.summary(target.symbol.attribute("__init__")) if project is not None: through = target.symbol.canonical_name.removeprefix("python.") - return self.known(project, through, arguments, keywords, everything, call, flows, state) + bound = self.receiver(call, project.name) + arguments = (*(self.deep(r, state) for r in bound), *arguments) + return self.known( + project, through, arguments, keywords, everything, call, flows, state, (), bound + ) symbol = target.symbol if not self.modelled(symbol): # ``get_conn().execute`` derived ``app.database.get_conn.execute``; what @@ -292,15 +303,33 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) captured = self.captured(call) - arguments = (*arguments, *(self.deep(value, state) for value in captured)) + bound = self.receiver(call, target.name) + arguments = ( + *(self.deep(r, state) for r in bound), + *arguments, + *(self.deep(value, state) for value in captured), + ) return self.known( - summary, target.name, arguments, keywords, everything, call, flows, state, captured + summary, target.name, arguments, keywords, everything, call, flows, state, captured, bound ) returned = self.returned_symbol(call) if returned is not None: return self.external(returned, everything, call, state, flows) return everything.join(state.get(call.callee, Taint.none())) + def receiver(self, call: Call, name: str) -> tuple[Value, ...]: + """The object a method call runs on, its implicit first parameter: the receiver + of ``obj.method(...)``, or the new object of ``Class(...)``.""" + + if "." not in name: + return () + callee = self.defs.get(call.callee) + if isinstance(callee, GetAttr): + return (callee.object,) + if name.endswith(".__init__") and isinstance(callee, Global | Symbol): + return (call.result,) + return () + def captured(self, call: Call) -> tuple[Value, ...]: """The values a nested callee captured, its implicit trailing parameters.""" @@ -372,6 +401,7 @@ def known( flows: list[TaintFlow], state: dict[Key, Taint], captured: tuple[Value, ...] = (), + receiver: tuple[Value, ...] = (), ) -> Taint: """Flows and result taint of a call to a function whose summary is known.""" @@ -379,7 +409,7 @@ def known( return everything spread = (*call.starred, *(value for _, value in call.keywords)) - values = (*call.arguments, *captured) + values = (*receiver, *call.arguments, *captured) def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: taint, witness = Taint.none(), None @@ -654,8 +684,9 @@ def propagate_taint( parameters: Mapping[int, Source] | None = None, project: SummaryIndex | None = None, heap: HeapFacts | None = None, + seeds: Mapping[HeapLocation, Taint] | None = None, ) -> TaintFacts: - problem = _TaintProblem(name, function, models, graph, summaries, parameters, project, heap) + problem = _TaintProblem(name, function, models, graph, summaries, parameters, project, heap, seeds) solution = solve(problem, cfg) taints: dict[Key, Taint] = {} flows: list[TaintFlow] = [] @@ -690,28 +721,80 @@ class TaintAnalysis(FunctionAnalysis[TaintFacts]): def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: graph = ctx.get(CallGraphAnalysis) models = ctx.get(SecurityModelAnalysis) + scopes, symbols = ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis) + instances = factory_instances(ctx.module, scopes, symbols, ctx.get(SummaryAnalysis), ctx.get(ProjectSummaries)) + routes = ctx.get(RegisteredRoutes) + + def sources_of(member: nodes.Function) -> Mapping[int, Source]: + return parameter_sources(member, ctx.module, models, scopes, symbols, instances, routes) + + ssa = ctx.get(SSAAnalysis, function) + heap = ctx.get(HeapAnalysis, function) + sources = sources_of(function) + # Only a method the framework calls, an entry point, starts with what its + # siblings stored; a method the project calls gets its ``self`` from the caller. + seeds = ( + self_seeds(function, ctx.module, ssa, heap, models, graph, ctx.get(SummaryAnalysis), sources_of) + if sources + else {} + ) return propagate_taint( graph.name_of(function), - ctx.get(SSAAnalysis, function), + ssa, ctx.get(CFGAnalysis, function), models, graph, ctx.get(SummaryAnalysis), - parameter_sources( - function, - ctx.module, - models, - ctx.get(ScopeAnalysis), - ctx.get(SymbolAnalysis), - factory_instances( - ctx.module, - ctx.get(ScopeAnalysis), - ctx.get(SymbolAnalysis), - ctx.get(SummaryAnalysis), - ctx.get(ProjectSummaries), - ), - ctx.get(RegisteredRoutes), - ), + sources, ctx.get(ProjectSummaries), - ctx.get(HeapAnalysis, function), + heap, + seeds, ) + + +def self_seeds( + function: nodes.Function, + module: nodes.Module, + ssa: FunctionIR, + heap: HeapFacts, + models: ModelTable, + graph: CallGraph, + summaries: SummaryTable, + sources_of: Callable[[nodes.Function], Mapping[int, Source]], +) -> dict[HeapLocation, Taint]: + """What the sibling methods of a method store into ``self`` from their own inputs: + the attributes ``self`` starts with when the framework, not the project, calls the + methods (``self.cmd = request.POST[...]`` in ``post``, read by ``get``).""" + + owner = next( + (s for s in module.body if isinstance(s, nodes.Class) and any(m is function for m in s.body)), + None, + ) + if owner is None or not ssa.parameters: + return {} + seeds: dict[HeapLocation, Taint] = {} + for sibling in owner.body: + if not isinstance(sibling, nodes.Function) or sibling is function: + continue + try: + summary = summaries.summary(graph.name_of(sibling)) + except KeyError: + continue + sources = sources_of(sibling) + for mutation in summary.mutations: + if mutation.parameter != 0: + continue + taint = Taint.none() + for index in mutation.dependencies: + source = sources.get(index) + if source is not None: + taint = taint.join(Taint(source.kinds, frozenset({source}))) + for symbol in mutation.externals: + external = models.source(symbol) + if external is not None: + taint = taint.join(Taint(external.kinds, frozenset({external}))) + if not taint: + continue + for location in heap.locations(ssa.parameters[0], mutation.field): + seeds[location] = seeds.get(location, Taint.none()).join(taint) + return seeds diff --git a/tests/test_self_state.py b/tests/test_self_state.py new file mode 100644 index 0000000..9bf245f --- /dev/null +++ b/tests/test_self_state.py @@ -0,0 +1,162 @@ +"""Acceptance tests for attribute state on ``self`` across methods (option 2, second +point of the consolidation). + +``self.cmd = argv`` in ``__init__`` then ``os.system(self.cmd)`` in ``run()`` was +invisible: each method saw its own ``self``. Three mechanisms close the gap. + +- The call graph knows the classes of a module: ``App(x)`` is a call to ``App.__init__`` + whose receiver is the new object, ``app.run()`` on an instance and ``self.helper()`` + inside a method are calls to the class's methods with the receiver as ``self``. +- A method's summary expresses what it reads from ``self`` as a dependency on its first + parameter, and callers map that parameter onto the receiver, contents included, so a + value stored by one method and read by another flows through the object, in one file + or across files through the project index. +- When the framework calls the methods, as for a Django view, a method starts with the + attributes of ``self`` seeded by what its sibling methods store there from their own + inputs, so ``self.cmd = request.POST[...]`` in ``post`` reaches ``os.system`` in ``get``. + +Expected to remain red until the call graph resolves instantiations and method calls and +the taint engine seeds ``self``. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.interprocedural import CallGraphAnalysis, KnownFunction, SummaryAnalysis +from coretrace_python.source import SourceManager + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +APP = ( + "import os\nimport sys\n\n" + "class App:\n" + " def __init__(self, cmd):\n self.cmd = cmd\n\n" + " def setup(self):\n self.cmd = input()\n\n" + " def run(self):\n os.system(self.cmd)\n\n" + " def helper(self):\n self.run()\n\n" +) + + +def hir(text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("s.py", text)) + + +def check(text: str, name: str = "s.py") -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source(name, text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, int, str | None, str]]: + return sorted((f.rule_id, f.span.start_line, f.function, f.metadata.get("through", "")) for f in findings) + + +@pytest.fixture(autouse=True) +def require_self_state() -> None: + graph = engine.build_manager(hir(APP + "def main():\n App(sys.argv[1]).run()\n")).get(CallGraphAnalysis) + if not any(isinstance(s.target, KnownFunction) and s.target.name == "App.__init__" for s in graph.sites("main")): + pytest.fail("self state across methods is not implemented yet") + + +# --------------------------------------------------------------------------- call graph + + +def test_instantiations_and_method_calls_resolve_to_the_class_methods() -> None: + graph = engine.build_manager(hir(APP + "def main(a: App):\n app = App(sys.argv[1])\n app.run()\n a.setup()\n")).get(CallGraphAnalysis) + + assert [s.target for s in graph.sites("main") if isinstance(s.target, KnownFunction)] == [ + KnownFunction("App.__init__"), + KnownFunction("App.run"), + KnownFunction("App.setup"), + ] + assert [s.target for s in graph.sites("App.helper")] == [KnownFunction("App.run")] + assert graph.callers("App.run") == frozenset({"main", "App.helper"}) + + +def test_unknown_classes_stay_unknown() -> None: + from coretrace_python.interprocedural import UnknownTarget + + graph = engine.build_manager(hir("def main(thing):\n thing.run()\n Other().go()\n")).get(CallGraphAnalysis) + assert all(isinstance(s.target, UnknownTarget) for s in graph.sites("main")) + + +# --------------------------------------------------------------------------- driven flows + + +def test_a_value_stored_by_init_reaches_a_sink_in_another_method() -> None: + findings = check(APP + "def main():\n app = App(sys.argv[1])\n app.run()\n") + assert rules(findings) == [("command-injection", 19, "main", "App.run")] + assert findings[0].metadata["sink_line"] == "12" + + +def test_a_value_stored_by_a_setter_method_reaches_a_later_method() -> None: + findings = check(APP + "def main():\n app = App('ls')\n app.setup()\n app.run()\n") + assert rules(findings) == [("command-injection", 20, "main", "App.run")] + + +def test_clean_instances_and_other_classes_are_not_confused() -> None: + assert check(APP + "def main():\n app = App('ls')\n app.run()\n") == () + assert ( + check( + APP + "class Other:\n def __init__(self):\n self.cmd = input()\n\n" + "def main():\n Other()\n App('ls').run()\n" + ) + == () + ) + + +def test_method_chains_go_through_self_calls() -> None: + findings = check(APP + "def main():\n App(input()).helper()\n") + assert rules(findings) == [("command-injection", 18, "main", "App.helper")] + + +def test_summaries_read_self_as_the_first_parameter() -> None: + table = engine.build_manager(hir(APP)).get(SummaryAnalysis) + + run = table.summary("App.run") + assert [(str(c.symbol), c.argument_dependencies) for c in run.external_calls] == [("python.os.system", (frozenset({0}),))] + init = table.summary("App.__init__") + assert [(m.parameter, m.field, m.dependencies) for m in init.mutations] == [(0, "attributes", frozenset({1}))] + + +def test_instances_cross_files_through_the_project_index(tmp_path: Path) -> None: + (tmp_path / "app.py").write_text(APP, encoding="utf-8") + (tmp_path / "main.py").write_text("import sys\nfrom app import App\n\ndef main():\n App(sys.argv[1]).run()\n", encoding="utf-8") + + findings = engine.analyze_project(tmp_path, [PLUGINS]).findings + + assert [(f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line, f.metadata.get("through")) for f in findings] == [ + ("command-injection", "main.py", 5, "app.App.run") + ] + + +# --------------------------------------------------------------------------- framework-driven methods + + +def test_sibling_methods_seed_self_for_entry_point_classes() -> None: + findings = check( + "import os\nfrom django.views import View\n\n" + "class Runner(View):\n" + " def post(self, request):\n self.cmd = request.POST['cmd']\n return 'stored'\n\n" + " def get(self, request):\n os.system(self.cmd)\n return 'ran'\n" + ) + assert rules(findings) == [("command-injection", 10, "get", "")] + assert findings[0].metadata["source_label"] == "http" + + +def test_seeding_needs_a_sibling_that_stores_input() -> None: + assert ( + check( + "import os\nfrom django.views import View\n\n" + "class Runner(View):\n" + " def post(self, request):\n self.cmd = 'ls'\n return 'stored'\n\n" + " def get(self, request):\n os.system(self.cmd)\n return 'ran'\n" + ) + == () + )