diff --git a/README.md b/README.md index 3c72534..b03179a 100644 --- a/README.md +++ b/README.md @@ -159,9 +159,10 @@ assignment, list, tuple and dict literals with `*` and `**` unpacking, f-strings `try`/`except`/`else`/`finally`, `await`, `yield`, `if`/`elif`/`else`, `while`, `for` with name or tuple targets, `break`, `continue` and `raise`, `from` clause included, conditional expressions and comprehensions, laid out as real branches and loops over a -synthetic local, set literals, chained assignments, assignments to `global` names, and -lambdas and nested function definitions as function values whose bodies are not analysed -yet, `del`, loop `else` clauses, and `match` over literal, singleton, capture, wildcard +synthetic local, set literals, chained assignments, assignments to `global` names, +lambdas and nested function definitions, analysed as functions of their own whose +captured variables are implicit parameters so taint flows through closures like through +any call, `del`, loop `else` clauses, and `match` over literal, singleton, capture, wildcard and or-patterns with guards, laid out as an `if` chain. Not yet: sequence, mapping and class patterns, `nonlocal` assignments, classes defined inside functions, and a conditional expression or comprehension inside a `while` condition. Blocks inside a `try` diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index b0fe63f..f8514c0 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -23,6 +23,7 @@ GetAttr, GetItem, Global, + MakeFunction, Symbol, Value, WithEnter, @@ -149,8 +150,10 @@ def resolve_targets( scopes: ScopeTable, known: frozenset[str], parameters: Mapping[Value, SymbolId] | None = None, + nested: frozenset[str] = frozenset(), ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]: - """Map every callee value of ``function`` to its target, and every symbol value.""" + """Map every callee value of ``function`` to its target, and every symbol value. + ``nested`` names the functions defined inside this one, ``outer.inner``.""" module = scopes.module_scope symbols = derive_symbols(function, parameters) @@ -159,6 +162,10 @@ def resolve_targets( } 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): binding = module.bindings.get(instruction.name) if ( @@ -178,7 +185,8 @@ def _annotated( scope = scopes.scope_for(function) enclosing = scope.parent if scope.parent is not None else scope.id found: dict[Value, SymbolId] = {} - for value, parameter in zip(ssa.parameters, function.parameters, strict=True): + # Captured variables follow the explicit parameters and carry no annotation. + for value, parameter in zip(ssa.parameters, function.parameters, strict=False): if parameter.annotation is None: continue symbol = table.resolve_expression(enclosing, parameter.annotation) @@ -218,7 +226,9 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph: unsupported.add(name) sites[name] = () continue - targets, symbols[name] = resolve_targets(ssa, scopes, known, _annotated(function, ssa, scopes, table)) + targets, symbols[name] = resolve_targets( + ssa, scopes, known, _annotated(function, ssa, scopes, table), frozenset(definitions) + ) found: list[CallSite] = [] for block in ssa.blocks: for instruction in block.instructions: diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index f052e56..e00b851 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -49,6 +49,7 @@ Global, Instruction, Jump, + MakeFunction, Phi, Return, SetAttr, @@ -310,6 +311,9 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: ) return everything | Dep(externals=frozenset({target.symbol})) if isinstance(target, KnownFunction): + 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) assert isinstance(target, UnknownTarget) return everything | state.get(call.callee, EMPTY) @@ -342,10 +346,12 @@ def mapped(indices: Dependencies) -> Dep: reached.location, call.location, ) + made = self.defs.get(call.callee) + values = (*call.arguments, *(made.captured if isinstance(made, MakeFunction) else ())) for mutation in callee.mutations: - if mutation.parameter < len(call.arguments): + if mutation.parameter < len(values): deps = mapped(mutation.dependencies) | Dep(externals=mutation.externals) - self.store(state, call.arguments[mutation.parameter], mutation.field, deps) + self.store(state, values[mutation.parameter], mutation.field, deps) return mapped(callee.return_dependencies) | Dep(externals=callee.return_externals) def record( diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index 30fe373..25a39e1 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -134,6 +134,9 @@ def expression(self, node: nodes.Expression) -> Value: if isinstance(node, nodes.Name): resolution = self.resolve(node.identifier) if resolution.kind is ResolutionKind.FREE: + # A captured variable is an implicit parameter of the nested function. + if node.identifier in self.parameters: + return self.parameters[node.identifier] self.fail(node, "closures are not supported yet") if resolution.kind is not ResolutionKind.LOCAL: return self.emit(Global(self.new_value(), node.span, node.identifier)) @@ -181,7 +184,9 @@ def expression(self, node: nodes.Expression) -> Value: elements, unpacked = self.spread(node.elements) return self.emit(BuildSet(self.new_value(), node.span, elements, unpacked)) if isinstance(node, nodes.Lambda): - return self.emit(MakeFunction(self.new_value(), node.span, "")) + synthesized = lambda_function(node) + captured = tuple(self.captured_value(name, node.span) for name in captured_names(synthesized, self.scopes)) + return self.emit(MakeFunction(self.new_value(), node.span, synthesized.name, captured)) if isinstance(node, nodes.Conditional | nodes.Comprehension): self.fail(node, "expression-level control flow must be laid out by the CFG builder") if isinstance(node, nodes.Await): @@ -256,7 +261,8 @@ def statement(self, node: nodes.Statement) -> None: for parameter in node.parameters: if parameter.default is not None: self.expression(parameter.default) - made = self.emit(MakeFunction(self.new_value(), node.span, node.name)) + captured = tuple(self.captured_value(name, node.span) for name in captured_names(node, self.scopes)) + made = self.emit(MakeFunction(self.new_value(), node.span, node.name, captured)) self.store(nodes.Name(node.name, node.span), made) return if isinstance(node, nodes.AugAssign): @@ -345,20 +351,29 @@ def enter_loop(self, source: BlockId, target: BlockId) -> None: iterable = self.expression(header.iterable) self.iterators[target] = self.emit(GetIter(self.new_value(), header.span, iterable)) + def captured_value(self, name: str, span: SourceSpan) -> Value: + """The current value of a variable a nested function captures.""" + + return self.expression(nodes.Name(name, span)) + def function(self, node: nodes.Function) -> FunctionIR: - parameter_values = tuple(self.new_value() for _ in node.parameters) + captured = captured_names(node, self.scopes) + parameter_values = tuple(self.new_value() for _ in (*node.parameters, *captured)) reassigned = _reassigned_parameters(node) self.parameters = { parameter.name: value - for parameter, value in zip(node.parameters, parameter_values, strict=True) + for parameter, value in zip(node.parameters, parameter_values, strict=False) if parameter.name not in reassigned } + # Captured variables come after the explicit parameters; they are never + # reassigned, an assignment would have made them local. + self.parameters.update(zip(captured, parameter_values[len(node.parameters) :], strict=True)) blocks: list[BasicBlock] = [] for cfg_block in self.cfg.blocks.values(): self.instructions = [] if cfg_block.id == self.cfg.entry: # Reassigned parameters live in locals so every block reads the same slot. - for parameter, value in zip(node.parameters, parameter_values, strict=True): + for parameter, value in zip(node.parameters, parameter_values, strict=False): if parameter.name in reassigned: self.emit_effect(StoreLocal(None, parameter.span, parameter.name, value)) for statement in cfg_block.statements: @@ -375,17 +390,51 @@ def function(self, node: nodes.Function) -> FunctionIR: def qualified_name(scopes: ScopeTable, function: nodes.Function) -> str: - """``Class.method`` for methods, the bare name for module-level functions.""" + """``Class.method`` for methods, ``outer.inner`` for nested functions, the bare name + for module-level functions; lambda and comprehension scopes keep identifier names.""" names = [function.name] scope = scopes.scope_for(function) parent = scopes.scope(scope.parent) if scope.parent else None while parent is not None and parent.kind is not ScopeKind.MODULE: - names.append(parent.name) + names.append(parent.name.strip("<>")) parent = scopes.scope(parent.parent) if parent.parent else None return ".".join(reversed(names)) +def lambda_function(node: nodes.Lambda) -> nodes.Function: + """A lambda as a function named after its position, returning its body.""" + + return nodes.Function( + f"lambda_{node.span.start_line}_{node.span.start_column}", + node.parameters, + (nodes.Return(node.body, node.span),), + False, + node.span, + ) + + +def captured_names(function: nodes.Function, scopes: ScopeTable) -> tuple[str, ...]: + """The enclosing-function variables ``function`` reads, sorted: its implicit + parameters after the explicit ones.""" + + scope = scopes.scope_for(function) + names: set[str] = set() + + def walk(node: Node) -> None: + if isinstance(node, nodes.Name): + names.add(node.identifier) + for child in children(node): + walk(child) + + for statement in function.body: + walk(statement) + for parameter in function.parameters: + if parameter.default is not None: + walk(parameter.default) + return tuple(sorted(n for n in names if scopes.resolve(scope.id, n).kind is ResolutionKind.FREE)) + + def _reassigned_parameters(function: nodes.Function) -> frozenset[str]: """Parameters that the function body assigns to, outside nested scopes.""" @@ -434,17 +483,39 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> FunctionIR: def analyzable_functions(module: nodes.Module) -> tuple[nodes.Function, ...]: - """Top-level functions and the methods of top-level classes, in source order.""" + """Top-level functions, the methods of top-level classes, and the functions and + lambdas nested inside them, in source order.""" functions: list[nodes.Function] = [] for statement in module.body: if isinstance(statement, nodes.Function): - functions.append(statement) + _collect(statement, functions) elif isinstance(statement, nodes.Class): - functions.extend(s for s in statement.body if isinstance(s, nodes.Function)) + for member in statement.body: + if isinstance(member, nodes.Function): + _collect(member, functions) return tuple(functions) +def _collect(function: nodes.Function, into: list[nodes.Function]) -> None: + into.append(function) + + def walk(node: Node) -> None: + if isinstance(node, nodes.Function): + _collect(node, into) + return + if isinstance(node, nodes.Lambda): + _collect(lambda_function(node), into) + return + if isinstance(node, nodes.Class): + return + for child in children(node): + walk(child) + + for statement in function.body: + walk(statement) + + class ModuleIRAnalysis(Analysis[ModuleIR]): """Assemble the PyIR of every top-level function of the module.""" diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index a1ccc88..86a2ca8 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -166,11 +166,14 @@ class BuildSet(Operands): @dataclass(frozen=True) class MakeFunction(Operands): - """A function value: a lambda or a nested definition, whose body is not lowered.""" + """A function value: a lambda or a nested definition. ``captured`` are the values of + the variables the body captures, at the definition point, in the order the nested + function takes them as implicit parameters after its explicit ones.""" result: Value location: SourceSpan name: str + captured: tuple[Value, ...] = () @dataclass(frozen=True) diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index 94d8124..190b7b7 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -94,7 +94,10 @@ def _instruction(instruction: Instruction) -> str: parts = [_value(v) for v in instruction.elements] + [f"*{_value(v)}" for v in instruction.unpacked] return f"{_value(instruction.result)} = build_set {', '.join(parts)}".rstrip() if isinstance(instruction, MakeFunction): - return f'{_value(instruction.result)} = make_function "{instruction.name}"' + text = f'{_value(instruction.result)} = make_function "{instruction.name}"' + if instruction.captured: + text += f" [{', '.join(_value(v) for v in instruction.captured)}]" + return text if isinstance(instruction, DelItem): return f"del_item {_value(instruction.object)}, {_value(instruction.key)}" if isinstance(instruction, DelAttr): diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index f8faf8a..434e528 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -38,6 +38,7 @@ SummaryTable, project_symbol, ) +from coretrace_python.ir.lowering import analyzable_functions from coretrace_python.ir.model import ( BasicBlock, Branch, @@ -54,6 +55,7 @@ GetIter, Instruction, Jump, + MakeFunction, Phi, SetAttr, SetItem, @@ -289,12 +291,22 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T return self.external(symbol, everything, call, state, flows) if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) - return self.known(summary, target.name, arguments, keywords, everything, call, flows, state) + captured = self.captured(call) + arguments = (*arguments, *(self.deep(value, state) for value in captured)) + return self.known( + summary, target.name, arguments, keywords, everything, call, flows, state, captured + ) 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 captured(self, call: Call) -> tuple[Value, ...]: + """The values a nested callee captured, its implicit trailing parameters.""" + + made = self.defs.get(call.callee) + return made.captured if isinstance(made, MakeFunction) else () + def modelled(self, symbol: SymbolId) -> bool: return ( self.models.sink(symbol) is not None @@ -359,6 +371,7 @@ def known( call: Call, flows: list[TaintFlow], state: dict[Key, Taint], + captured: tuple[Value, ...] = (), ) -> Taint: """Flows and result taint of a call to a function whose summary is known.""" @@ -366,6 +379,7 @@ def known( return everything spread = (*call.starred, *(value for _, value in call.keywords)) + values = (*call.arguments, *captured) def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: taint, witness = Taint.none(), None @@ -373,7 +387,7 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: positional = index < len(arguments) part = arguments[index] if positional else keywords if part and witness is None and (positional or spread): - witness = call.arguments[index] if positional else spread[0] + witness = values[index] if positional else spread[0] taint = taint.join(part) return taint, witness @@ -387,13 +401,13 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: if witness is not None: self.report(flows, sink, taint, witness, call, through, reached.location, position) for mutation in summary.mutations: - if mutation.parameter < len(call.arguments): + if mutation.parameter < len(values): stored = mapped(mutation.dependencies)[0] for symbol in sorted(mutation.externals, key=str): stored_source = self.models.source(symbol) if stored_source is not None: stored = stored.join(Taint(stored_source.kinds, frozenset({stored_source}))) - self.store(state, call.arguments[mutation.parameter], mutation.field, stored) + self.store(state, values[mutation.parameter], mutation.field, stored) result = mapped(summary.return_dependencies)[0] for symbol in sorted(summary.return_externals, key=str): returned_source = self.models.source(symbol) @@ -477,6 +491,61 @@ def factory_instances( return found +def local_instances(function: nodes.Function, scopes: ScopeTable, symbols: SymbolTable) -> Instances: + """Locals of ``function`` bound to the result of calling a resolvable symbol.""" + + scope = scopes.scope_for(function).id + found: dict[str, tuple[SymbolId, ...]] = {} + for statement in function.body: + if ( + isinstance(statement, nodes.Assign) + and isinstance(statement.target, nodes.Name) + and isinstance(statement.value, nodes.Call) + ): + symbol = symbols.resolve_expression(scope, statement.value.callee) + if symbol is not None: + found[statement.target.identifier] = (symbol,) + return found + + +def _enclosing(module: nodes.Module, function: nodes.Function) -> nodes.Function | None: + """The function whose body defines ``function``, if it is nested.""" + + def search(body: tuple[nodes.Statement, ...], parent: nodes.Function | None) -> nodes.Function | None: + for statement in body: + if isinstance(statement, nodes.Function): + if statement.span == function.span: + return parent + found = search(statement.body, statement) + if found is not None: + return found + elif isinstance(statement, nodes.Class): + found = search(statement.body, None) + if found is not None: + return found + return None + + found = search(module.body, None) + if found is not None: + return found + # Lambdas are synthesized functions: their enclosing function is the innermost + # analysable function whose span contains theirs. + innermost: nodes.Function | None = None + for candidate in analyzable_functions(module): + if candidate.span != function.span and _contains(candidate.span, function.span): + innermost = candidate + return innermost + + +def _contains(outer: SourceSpan, inner: SourceSpan) -> bool: + if outer.source_id != inner.source_id or outer.end_line is None or inner.end_line is None: + return False + return (outer.start_line, outer.start_column) <= (inner.start_line, inner.start_column) and ( + inner.end_line, + inner.end_column, + ) <= (outer.end_line, outer.end_column) + + def _instance_symbols(expression: nodes.Expression, instances: Instances) -> tuple[SymbolId, ...]: """The symbols an expression rooted at a factory instance may denote.""" @@ -540,6 +609,10 @@ def parameter_sources( None, ) sources: dict[int, Source] = {} + enclosing_function = _enclosing(module, function) + if enclosing_function is not None: + # ``app = Flask(__name__)`` inside ``create_app``: routes defined there resolve. + instances = {**(instances or {}), **local_instances(enclosing_function, scopes, symbols)} entry = entry_point_of(function, models, scopes, symbols, owner, instances) if entry is None and routes: qualified = function.name if owner is None else f"{owner.name}.{function.name}" diff --git a/tests/test_closures.py b/tests/test_closures.py new file mode 100644 index 0000000..d258d13 --- /dev/null +++ b/tests/test_closures.py @@ -0,0 +1,188 @@ +"""Acceptance tests for the bodies of nested functions and lambdas (option 2, first +point of the consolidation agreed with Hugo). + +Until now a nested ``def`` or a ``lambda`` was a ``MakeFunction`` value whose body was +never analysed: a sink inside a callback was invisible and a call to it conservatively +tainted its result. Now every nested function and lambda inside an analysable function +is an analysable function of its own, named after its enclosing functions +(``outer.inner``, ``outer.lambda_L_C``). The variables it captures are implicit +parameters appended to its explicit ones; ``MakeFunction`` carries the captured values +at the definition point, and a call to a nested function maps the callee's captured +parameters onto them, so summaries and taint flow through closures like through any +known function. ``nonlocal`` assignments stay unsupported. + +Expected to remain red until ``captured_names``, ``MakeFunction.captured`` and nested +analysable functions exist. +""" + +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 SummaryAnalysis +from coretrace_python.ir.lowering import analyzable_functions, lower_module +from coretrace_python.ir.model import MakeFunction +from coretrace_python.ir.printer import format_module +from coretrace_python.semantic.scopes import analyze_scopes +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager + +try: + from coretrace_python.ir.lowering import captured_names +except ImportError as error: # pragma: no cover - red until closures land + MISSING: Exception | None = error +else: + MISSING = None + if "captured" not in MakeFunction.__dataclass_fields__: + MISSING = AttributeError("MakeFunction has no captured values") + + +@pytest.fixture(autouse=True) +def require_closures() -> None: + if MISSING is not None: + pytest.fail(f"closure bodies are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def hir(text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("c.py", text)) + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("c.py", 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) + + +NESTED = ( + "import os\n\n" + "def outer(prefix):\n" + " cmd = input()\n" + " def run(suffix):\n" + " os.system(prefix + cmd + suffix)\n" + " run('!')\n" + " key = lambda item: item.lower()\n" + " return key\n" +) + + +# --------------------------------------------------------------------------- structure + + +def test_nested_functions_and_lambdas_are_analysable_functions() -> None: + module = hir(NESTED) + + names = [f.name for f in analyzable_functions(module)] + + assert names == ["outer", "run", "lambda_8_11"] + functions = lower_module(module).functions + assert [f.name for f in functions] == ["outer", "outer.run", "outer.lambda_8_11"] + + +def test_captured_variables_become_implicit_parameters() -> None: + module = hir(NESTED) + scopes = analyze_scopes(module) + outer = module.body[1] + assert isinstance(outer, nodes.Function) + run = next(s for s in outer.body if isinstance(s, nodes.Function)) + + assert captured_names(run, scopes) == ("cmd", "prefix") + assert captured_names(outer, scopes) == () + + functions = {f.name: f for f in lower_module(module).functions} + assert len(functions["outer.run"].parameters) == 3 + assert len(functions["outer.lambda_8_11"].parameters) == 1 + text = format_module(lower_module(module)) + assert "func @outer.run(%0, %1, %2) {" in text + assert 'make_function "run"' in text and "[" in text.split('make_function "run"')[1].split("\n")[0] + + +def test_make_function_carries_the_captured_values() -> None: + (outer, *_) = lower_module(hir(NESTED)).functions + made = [i for b in outer.blocks for i in b.instructions if isinstance(i, MakeFunction)] + + assert [(m.name, len(m.captured)) for m in made] == [("run", 2), ("lambda_8_11", 0)] + + +def test_nonlocal_assignments_are_still_reported() -> None: + findings = check("def outer():\n n = 0\n def bump():\n nonlocal n\n n = n + 1\n bump()\n return n\n") + assert [(f.rule_id, f.function) for f in findings] == [("unsupported-syntax", "bump")] + + +# --------------------------------------------------------------------------- taint through closures + + +def test_a_sink_inside_a_closure_is_reached_through_its_call() -> None: + findings = check(NESTED) + assert rules(findings) == [("command-injection", 7, "outer", "outer.run")] + assert findings[0].metadata["sink_line"] == "6" + + +def test_explicit_arguments_and_lambdas_flow_too() -> None: + assert rules(check("import os\n\ndef outer():\n def run(c):\n os.system(c)\n run(input())\n")) == [ + ("command-injection", 6, "outer", "outer.run") + ] + assert rules(check("import os\n\ndef outer():\n handler = lambda c: os.system(c)\n handler(input())\n")) == [ + ("command-injection", 5, "outer", "outer.lambda_4_15") + ] + assert rules(check("import os\n\ndef outer():\n cmd = input()\n fire = lambda: os.system(cmd)\n fire()\n")) == [ + ("command-injection", 6, "outer", "outer.lambda_5_12") + ] + + +def test_uncalled_and_clean_closures_stay_silent() -> None: + assert check("import os\n\ndef outer():\n cmd = input()\n def run():\n os.system(cmd)\n return run\n") == () + assert check("import os\n\ndef outer():\n def run(c):\n os.system('ls')\n run(input())\n") == () + + +def test_closures_compose_through_summaries() -> None: + manager = engine.build_manager(hir("import os\n\ndef outer(cmd):\n def run():\n os.system(cmd)\n run()\n")) + table = manager.get(SummaryAnalysis) + + run = table.summary("outer.run") + assert run.parameters == 1 and [str(c.symbol) for c in run.external_calls] == ["python.os.system"] + assert run.external_calls[0].argument_dependencies == (frozenset({0}),) + outer = table.summary("outer") + assert [(str(c.symbol), c.argument_dependencies) for c in outer.external_calls] == [("python.os.system", (frozenset({0}),))] + + +def test_routes_defined_inside_an_application_factory_are_entry_points() -> None: + findings = check( + "import os\nfrom flask import Flask, request\n\n" + "def create_app():\n" + " app = Flask(__name__)\n\n" + " @app.route('/ping/')\n" + " def ping(host):\n" + " os.system('ping ' + host)\n" + " return 'ok'\n\n" + " return app\n" + ) + assert rules(findings) == [("command-injection", 9, "ping", "")] + assert findings[0].metadata["source"] == "python.flask.Flask.route" + + +def test_coverage_counts_nested_functions() -> None: + analysis = engine.analyze_file(SourceManager().add_source("c.py", NESTED), [PLUGINS]) + assert analysis.coverage.summary() == "coverage: 1/1 files, 3/3 functions" + + +def test_call_graph_names_nested_functions_by_their_enclosing_ones() -> None: + from coretrace_python.interprocedural import CallGraphAnalysis, KnownFunction + + manager = engine.build_manager(hir(NESTED)) + graph = manager.get(CallGraphAnalysis) + + assert graph.functions == ("outer", "outer.run", "outer.lambda_8_11") + assert [s.target for s in graph.sites("outer") if isinstance(s.target, KnownFunction)] == [KnownFunction("outer.run")] + assert SymbolId("python.c.outer.run") is not None diff --git a/tests/test_syntax_control_flow.py b/tests/test_syntax_control_flow.py index 6604287..4a25cab 100644 --- a/tests/test_syntax_control_flow.py +++ b/tests/test_syntax_control_flow.py @@ -149,9 +149,9 @@ def test_lambdas_have_their_own_scope_and_lower_to_function_values() -> None: assert isinstance(returned.value.keywords[0].value, Lambda) text = printed("def f(items, k):\n return sorted(items, key=lambda x: x + k)\n") - assert 'make_function ""' in text + assert 'make_function "lambda_2_30" [%1]' in text (made,) = [i for b in lower("def f():\n return lambda: 1\n").blocks for i in b.instructions if isinstance(i, MakeFunction)] - assert made.name == "" + assert made.name == "lambda_2_12" def test_calling_a_lambda_is_conservative() -> None: @@ -160,7 +160,7 @@ def test_calling_a_lambda_is_conservative() -> None: def test_nested_functions_become_values_and_the_outer_function_is_analysed() -> None: module = hir("def outer():\n def inner(x):\n return x\n os.system(inner(input()))\n") - assert [f.name for f in analyzable_functions(module)] == ["outer"] + assert [f.name for f in analyzable_functions(module)] == ["outer", "inner"] text = printed("def outer():\n def inner(x):\n return x\n return inner\n") assert 'make_function "inner"' in text and 'store_local "inner"' in text