From c21b92d135553e751c57457f2b66fd3029011bd4 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 05:59:53 +0200 Subject: [PATCH 1/4] feat(code-injection): report attributes chosen by input and then called Part of #240: getattr(obj, name)(...), one finding per invocation. --- docs/plugins.md | 6 +- docs/usage.md | 2 +- .../models/python_stdlib/python_stdlib.py | 9 + .../security/code_injection/code_injection.py | 26 +- src/coretrace_python/cache.py | 5 +- .../interprocedural/summaries.py | 18 +- src/coretrace_python/ir/defuse.py | 27 +- src/coretrace_python/taint/engine.py | 84 ++++-- src/coretrace_python/taint/models.py | 24 +- tests/test_attribute_call_sinks.py | 271 ++++++++++++++++++ tests/test_sink_called_result.py | 66 +++++ 11 files changed, 506 insertions(+), 32 deletions(-) create mode 100644 tests/test_attribute_call_sinks.py create mode 100644 tests/test_sink_called_result.py diff --git a/docs/plugins.md b/docs/plugins.md index 7e64e04..757e318 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -169,7 +169,7 @@ class BottleModels(ModelPlugin): | Model | Declares | |---|---| | `Source(symbol, label, kinds=ALL)` | A call or attribute whose value is attacker-controlled; `label` names it in messages (`http`, `stdin`, `argv`, `http-response`). `kinds` restricts what it can inject. Text input, such as a form field, a header or a URL parameter, carries `TEXT_KINDS`, every kind but `NOSQL`: a string cannot hold a query operator. | -| `Sink(symbol, kinds, positions=(), keywords=())` | A call dangerous for `kinds`. `positions` limits a kind to the given argument indexes, and `keywords` to the given keyword names; a restricted kind reaches only the arguments listed there. SQL sinks read the statement only, and the SSRF sinks of HTTP clients read the URL, by position or as `url=`. A starred argument or `**kwargs` reaches no restricted kind. | +| `Sink(symbol, kinds, positions=(), keywords=(), selects=False, requires_called_result=NONE)` | A call dangerous for `kinds`. `positions` limits a kind to the given argument indexes, and `keywords` to the given keyword names; a restricted kind reaches only the arguments listed there. SQL sinks read the statement only, and the SSRF sinks of HTTP clients read the URL, by position or as `url=`. A starred argument or `**kwargs` reaches no restricted kind. `selects` marks an argument that chooses by name the code that runs (a module, a file, a directory, an attribute): only a check limiting it to constants proves it. `requires_called_result` lists the kinds that apply only when the call's result is itself called, as for `getattr`'s name: the engine reports them at each invocation of the result, the other kinds at the call. | | `Sanitizer(symbol, kinds)` | A call whose result is safe for `kinds`, including when the call is inside a project function the data goes through. | | `EntryPoint(symbol, label, kinds=ALL, inputs=None)` | A decorator or a class base whose functions receive their parameters as `label` input, such as `flask.Flask.route`; or a project function or class named by its own symbol (`python.app.handlers.main`), as an integration reading a deployment declaration names a Lambda handler. `inputs` limits the parameters that are input to those positions, after `self` for a method: a handler's event, not its context. | | `TypedParameter(symbol, label, kinds=ALL)` | A class whose annotated parameters carry `label` input, such as Django's `HttpRequest`. | @@ -187,7 +187,9 @@ class BottleModels(ModelPlugin): Two plugins may describe the same symbol. An identical model is registered once; two `Sink` models merge their kinds, positions and keywords, so a plugin can add `SQL` to a sink another plugin declared for `COMMAND`; any other difference is a conflict that stops -the analysis with a `ModelError` naming both plugins. +the analysis with a `ModelError` naming both plugins. A merged sink `selects` when either +does, and a kind requires the called result only when every declaration carrying it +does: an unconditional declaration is never weakened, whatever the order. A plugin may also read models from the project it analyses: `project_models(self, root)` returns them, from a file at `root` for instance. It is called once per directory check, diff --git a/docs/usage.md b/docs/usage.md index 5c49577..aac0310 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -121,7 +121,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| | `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. The arguments after a script or code are the script's own: options of unknown meaning, reported `medium` (a hotspot after an option of the script), refuted behind a constant prefix. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. Every case keeps the rule id `command-injection`, so suppressions and baselines match whatever the case. When the rule reads the command element by element and establishes what the input is there, the finding's `injection` metadata (in the JSON report, and as a property of its SARIF result) says so: `command` when the input runs or chooses what runs (a shell string, the program, an operand run as a command, the value of an option that runs it); `option` when it can only reach the program as an option or as the value of an option not established to run it, and the message is then titled `Option injection`. When that cannot be established (an option whose effect is to review, an unknown program or element before the input, a command that is not a list or tuple display, a sink such as `os.system` the rule does not read element by element), the metadata is absent and the title stays `Command injection`. | -| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. | +| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. Also the name `getattr(obj, name)` receives, when the attribute it selects is called: the finding, titled `Untrusted attribute call`, is placed at each call of the result, one per call, with the lines of the selection and of the call in its metadata (`selected_line`, `invoked_line`), and through project functions at the call that reaches them. What runs depends on what the object exposes, so it is a hotspot, on a module too and behind a constant prefix; a check proves the name only when it limits the value `getattr` received to constants, wherever it stands before the call, and a check of a name rebound since proves nothing. A `getattr` whose result is only read is not reported, nor, in this version, one whose result is returned to the caller, stored in an attribute or a container, or passed on before it is called, nor `operator.attrgetter`, `operator.methodcaller`, `obj.__getattribute__(name)` or `vars(obj)[name]`; checks made inside a project function are not seen from its callers, as for every sink. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | | `path-traversal` | A file system path (`open`, `send_file`, `os.remove`, …). | diff --git a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py index e9a3afd..496b325 100644 --- a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py +++ b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py @@ -80,6 +80,15 @@ class PythonStdlibModels(ModelPlugin): Sink(_sym("sys.path.append"), TaintKind.CODE, ((TaintKind.CODE, (0,)),), selects=True), Sink(_sym("sys.path.extend"), TaintKind.CODE, ((TaintKind.CODE, (0,)),), selects=True), Sink(_sym("site.addsitedir"), TaintKind.CODE, ((TaintKind.CODE, (0,)),), ((TaintKind.CODE, ("sitedir",)),), selects=True), + # The name of the attribute ``getattr`` selects, when what it selects is called; + # the object and the default are data (#240). + Sink( + _sym("builtins.getattr"), + TaintKind.CODE, + ((TaintKind.CODE, (1,)),), + selects=True, + requires_called_result=TaintKind.CODE, + ), Sink(_sym("builtins.open"), TaintKind.PATH), Sink(_sym("os.remove"), TaintKind.PATH), Sink(_sym("os.unlink"), TaintKind.PATH), diff --git a/src/coretrace_python/bundled/security/code_injection/code_injection.py b/src/coretrace_python/bundled/security/code_injection/code_injection.py index 6a8280a..247ab54 100644 --- a/src/coretrace_python/bundled/security/code_injection/code_injection.py +++ b/src/coretrace_python/bundled/security/code_injection/code_injection.py @@ -1,8 +1,9 @@ """Code injection: attacker-controlled input reaching a CODE sink, such as the code ``eval`` or ``exec`` runs, the module ``importlib.import_module``, ``__import__`` or -``runpy.run_module`` imports, the file ``runpy.run_path`` runs, or the directory -``sys.path`` and ``site.addsitedir`` load code from. ``dangerous-eval`` reports every -``eval`` or ``exec``; this rule reports the calls attacker input reaches. +``runpy.run_module`` imports, the file ``runpy.run_path`` runs, the directory +``sys.path`` and ``site.addsitedir`` load code from, or the attribute ``getattr`` selects +when it is called. ``dangerous-eval`` reports every ``eval`` or ``exec``; this rule reports +the calls attacker input reaches. An import runs the top-level code of a module the input chooses among those that can be found, and a script path runs the file the input chooses: execution is possible, and @@ -11,7 +12,9 @@ modules of that package: a hotspot. A directory added to ``sys.path`` runs code only when a later import resolves there: a hotspot. ``site.addsitedir`` runs the import lines of the directory's ``.pth`` files during the call: a vulnerability, exploitable when the -directory holds files the attacker controls, as the finding states.""" +directory holds files the attacker controls, as the finding states. The attribute +``getattr`` selects runs at each call of its result, where the finding is placed: what +runs depends on what the object exposes, a hotspot.""" from __future__ import annotations @@ -34,6 +37,7 @@ _RUN_PATH = SymbolId("python.runpy.run_path") _IMPORT_PATHS = frozenset(SymbolId(f"python.sys.path.{name}") for name in ("insert", "append", "extend")) _SITE_DIRECTORY = SymbolId("python.site.addsitedir") +_GETATTR = SymbolId("python.builtins.getattr") class CodeInjectionPlugin(TaintDetector): @@ -59,6 +63,20 @@ def assess( "import resolves to a module the attacker placed there", ) return Assessment(verdict, self.severity, "Untrusted import path") + if symbol == _GETATTR: + if verdict.status is Status.VULNERABILITY: + verdict = Verdict( + flow, + Status.HOTSPOT, + "the input chooses the attribute called: what runs depends on what the object exposes", + ) + lines: dict[str, str] = {} + if flow.invoked_at is not None and flow.sink_location is not None: + lines = { + "selected_line": str(flow.sink_location.start_line), + "invoked_line": str(flow.invoked_at.start_line), + } + return Assessment(verdict, self.severity, "Untrusted attribute call", lines) if symbol == _SITE_DIRECTORY: if verdict.status is Status.VULNERABILITY: verdict = Verdict( diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 2414194..6bf2d39 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -42,7 +42,7 @@ from coretrace_python.semantic.symbols import SymbolId from coretrace_python.source import FileLocation, Location, SourceId, SourceSpan -CACHE_FORMAT = 14 +CACHE_FORMAT = 15 @dataclass(frozen=True) @@ -253,11 +253,13 @@ def _encode_call(call: ExternalCall) -> dict[str, Any]: "given": _encode_arguments(call.arguments), "argument_cleared": [_encode_cleared(c) for c in call.argument_cleared], "keyword_cleared": [_encode_cleared(c) for c in call.keyword_cleared], + "invoked_at": None if call.invoked_at is None else _encode_span(call.invoked_at), } def _decode_call(data: Mapping[str, Any]) -> ExternalCall: site = data["call_site"] + invoked_at = data["invoked_at"] return ExternalCall( SymbolId(_string(data["symbol"])), tuple(_indices(deps) for deps in data["arguments"]), @@ -267,6 +269,7 @@ def _decode_call(data: Mapping[str, Any]) -> ExternalCall: _decode_arguments(data["given"]), tuple(_decode_cleared(c) for c in data["argument_cleared"]), tuple(_decode_cleared(c) for c in data["keyword_cleared"]), + None if invoked_at is None else _decode_span(invoked_at), ) diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 0e2b770..5e807fe 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -41,7 +41,7 @@ UnknownTarget, _is_staticmethod, ) -from coretrace_python.ir.defuse import DefUse, def_use +from coretrace_python.ir.defuse import DefUse, def_use, invocations from coretrace_python.ir.model import ( BasicBlock, Branch, @@ -301,6 +301,10 @@ class ExternalCall: # What was cleared on the way to each argument and keyword, in the same order. argument_cleared: tuple[Cleared, ...] = () keyword_cleared: tuple[Cleared, ...] = () + # The call of the result this record is for, ``None`` for the call itself: one record + # per invocation, the call at ``location`` (``getattr``'s) staying the selection, so + # that a consumer applies a sink's conditional kinds once at each invocation (#240). + invoked_at: SourceSpan | None = None def positional(self) -> Iterable[tuple[Dependencies, Cleared]]: """Each positional argument's dependencies, with what was cleared on their way.""" @@ -398,7 +402,7 @@ def summary(self, name: str) -> FunctionSummary: Key = Value | HeapLocation State = Mapping[Key, Dep] -_CallKey = tuple[SymbolId, SourceSpan, SourceSpan | None] +_CallKey = tuple[SymbolId, SourceSpan, SourceSpan | None, SourceSpan | None] class _DependenceProblem(DataflowProblem[State]): @@ -565,6 +569,11 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: return self.known(project, arguments, keywords, everything, call, state, bound) given = self.graph.arguments_at(self.name, call.location) self.record(target.symbol, arguments, named, call.location, None, given) + # Where its result is called: facts the summary keeps whatever the models; + # the consumer decides which kinds of a sink apply there (#240). + if call.result is not None: + for invocation in invocations(self.function, self.uses, call.result): + self.record(target.symbol, arguments, named, call.location, None, given, invocation.location) self.record_filters(target.symbol, call, given, state) cleared = everything.clearing(self.clearing.clears(target.symbol, given)) return cleared | Dep(externals=frozenset({target.symbol})) @@ -650,6 +659,7 @@ def mapped(indices: Dependencies, cleared: Cleared = ()) -> Dep: reached.location, call.location, reached.arguments, + reached.invoked_at, ) made = self.defs.get(call.callee) values = (*receiver, *call.arguments, *(made.captured if isinstance(made, MakeFunction) else ())) @@ -692,8 +702,9 @@ def record( location: SourceSpan, call_site: SourceSpan | None, given: Arguments, + invoked_at: SourceSpan | None = None, ) -> None: - key = (symbol, location, call_site) + key = (symbol, location, call_site, invoked_at) previous = self.external.get(key) if previous is not None: arguments = tuple( @@ -713,6 +724,7 @@ def record( given, tuple(a.cleared for a in arguments), tuple(deps.cleared for _, deps in keywords), + invoked_at, ) diff --git a/src/coretrace_python/ir/defuse.py b/src/coretrace_python/ir/defuse.py index 24187bf..490fe00 100644 --- a/src/coretrace_python/ir/defuse.py +++ b/src/coretrace_python/ir/defuse.py @@ -10,7 +10,7 @@ from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis from coretrace_python.cfg import BlockId from coretrace_python.hir import nodes -from coretrace_python.ir.model import ForNext, FunctionIR, Value +from coretrace_python.ir.model import Call, ForNext, FunctionIR, Phi, Value from coretrace_python.ir.ssa import SSAAnalysis @@ -65,6 +65,31 @@ def def_use(function: FunctionIR) -> DefUse: return DefUse(definitions, {value: tuple(found) for value, found in uses.items()}) +def invocations(function: FunctionIR, uses: DefUse, value: Value) -> tuple[Call, ...]: + """The calls of ``function`` whose callee is ``value``, directly or through the phis + that merge it on branches and loops, each once, in source order: where a callable + value is invoked. A value stored, returned or passed on is not followed.""" + + blocks = {block.id: block for block in function.blocks} + found: dict[int, Call] = {} + seen = {value} + pending = [value] + while pending: + current = pending.pop() + for use in uses.uses(current): + if use.index is None: + continue + instruction = blocks[use.block].instructions[use.index] + if isinstance(instruction, Phi) and instruction.result is not None and instruction.result not in seen: + seen.add(instruction.result) + pending.append(instruction.result) + elif isinstance(instruction, Call) and instruction.callee == current: + found[id(instruction)] = instruction + return tuple( + sorted(found.values(), key=lambda call: (call.location.start_line, call.location.start_column)) + ) + + class DefUseAnalysis(FunctionAnalysis[DefUse]): name: ClassVar[str] = "ir.defuse" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis}) diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index ac03a94..efca2a1 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -48,7 +48,7 @@ cleared_by, project_symbol, ) -from coretrace_python.ir.defuse import DefUse, def_use +from coretrace_python.ir.defuse import DefUse, def_use, invocations from coretrace_python.ir.lowering import analyzable_functions, is_body_function from coretrace_python.ir.model import ( BasicBlock, @@ -119,7 +119,11 @@ 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.""" + inside a known callee, ``through`` names it and ``sink_location`` points at the sink. + A sink kind that requires its call's result to be called (``getattr``'s name) gives one + flow per invocation: ``invoked_at`` is that invocation, ``sink_location`` the call that + selected what it runs, and ``location`` the invocation itself in the analysed function, + so that the checks made before it are judged (#240).""" source: Source sink: Sink @@ -134,6 +138,7 @@ class TaintFlow: # (None, None) unpacked with ``*`` or ``**``; ``f(x, key=x)`` passes it twice. The # keyword ``KEYWORD_NAMES`` says the value's keys are the keyword names of the call. passed_as: frozenset[tuple[int | None, str | None]] = frozenset() + invoked_at: SourceSpan | None = None Key = Value | HeapLocation @@ -529,19 +534,17 @@ def external( given = self.graph.arguments_at(self.name, call.location) sink = self.guarded(self.models.sink(symbol), given) if sink is not None: - for position, argument in enumerate(call.arguments): - self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, position, given) - for argument in call.starred: - self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, None, given) - for name, argument in call.keywords: - self.report( - flows, sink, self.rendered(argument, state), argument, call, None, None, None, given, keyword=name - ) - if name is None: - # The keys of a ``**mapping`` are keyword names of the call: only an - # advisory asks who chooses a name, no other sink reads one. - keys = self.keyed(argument, state).without(~TaintKind.ADVISORY) - self.report(flows, sink, keys, argument, call, None, None, None, given, keyword=KEYWORD_NAMES) + # The unconditional kinds at the call; those that require its result to be + # called at each invocation of it, where the flow is anchored (#240). + selection = sink.applying(invoked=False) + if selection is not None: + self.report_arguments(flows, selection, call, call, None, None, state, given) + invoked = sink.applying(invoked=True) + if invoked is not None and call.result is not None: + for invocation in invocations(self.function, self.uses, call.result): + self.report_arguments( + flows, invoked, call, invocation, call.location, invocation.location, state, given + ) self.template_filters(symbol, call, given, state, flows) cleared = self.clearing.clears(symbol, given) if cleared: @@ -553,6 +556,44 @@ def external( def uses(self) -> DefUse: return def_use(self.function) + def report_arguments( + self, + flows: list[TaintFlow], + sink: Sink, + call: Call, + anchor: Call, + sink_location: SourceSpan | None, + invoked_at: SourceSpan | None, + state: Mapping[Key, Taint], + given: Arguments, + ) -> None: + """The flows of ``call``'s arguments into ``sink``, anchored at ``anchor``: the call + itself, or an invocation of its result.""" + + for position, argument in enumerate(call.arguments): + self.report( + flows, sink, self.rendered(argument, state), argument, anchor, None, sink_location, position, given, + invoked_at=invoked_at, + ) + for argument in call.starred: + self.report( + flows, sink, self.rendered(argument, state), argument, anchor, None, sink_location, None, given, + invoked_at=invoked_at, + ) + for name, argument in call.keywords: + self.report( + flows, sink, self.rendered(argument, state), argument, anchor, None, sink_location, None, given, + keyword=name, invoked_at=invoked_at, + ) + if name is None: + # The keys of a ``**mapping`` are keyword names of the call: only an + # advisory asks who chooses a name, no other sink reads one. + keys = self.keyed(argument, state).without(~TaintKind.ADVISORY) + self.report( + flows, sink, keys, argument, anchor, None, sink_location, None, given, + keyword=KEYWORD_NAMES, invoked_at=invoked_at, + ) + def template_filters( self, symbol: SymbolId, call: Call, given: Arguments, state: Mapping[Key, Taint], flows: list[TaintFlow] ) -> None: @@ -673,17 +714,24 @@ def mapped(deps: frozenset[int], cleared: Cleared = ()) -> tuple[Taint, Value | for reached in summary.external_calls: sink = self.guarded(self.models.sink(reached.symbol), reached.arguments) + # A record of an invocation carries the kinds that require it, the record of + # the call itself the others (#240). + sink = None if sink is None else sink.applying(invoked=reached.invoked_at is not None) if sink is None: continue for position, (deps, cleared) in enumerate(reached.positional()): taint, witness = mapped(deps, cleared) if witness is not None: - self.report(flows, sink, taint, witness, call, through, reached.location, position, reached.arguments) + self.report( + flows, sink, taint, witness, call, through, reached.location, position, reached.arguments, + invoked_at=reached.invoked_at, + ) for name, deps, cleared in reached.named(): taint, witness = mapped(deps, cleared) if witness is not None: self.report( - flows, sink, taint, witness, call, through, reached.location, None, reached.arguments, keyword=name + flows, sink, taint, witness, call, through, reached.location, None, reached.arguments, + keyword=name, invoked_at=reached.invoked_at, ) for mutation in summary.mutations: if mutation.parameter < len(values): @@ -712,6 +760,7 @@ def report( position: int | None = None, arguments: Arguments | None = None, keyword: str | None = None, + invoked_at: SourceSpan | None = None, ) -> None: reaching = taint.kinds & sink.kinds_at(position, keyword) if not reaching: @@ -733,6 +782,7 @@ def report( call.location if sink_location is None else sink_location, arguments, frozenset({(position, keyword)}), + invoked_at, ) ) diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index f9cdd20..515d8ce 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -11,7 +11,7 @@ import hashlib import re from collections.abc import Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from enum import Flag, auto from types import MappingProxyType from typing import ClassVar @@ -68,13 +68,17 @@ class Sink: ``requests.get`` is its first argument or ``url=``, its body is no destination. ``selects`` marks a sink whose argument chooses by name the code that runs, a module, a file or a directory code is loaded from: only a check that limits the value to - constants proves it, not a check of its shape (#240).""" + constants proves it, not a check of its shape (#240). ``requires_called_result`` holds + the kinds that apply only when the call's result is itself called, ``getattr``'s name + choosing the attribute the call runs: the engine applies them at each invocation of + the result, the other kinds at the call (#240).""" symbol: SymbolId kinds: TaintKind positions: tuple[tuple[TaintKind, tuple[int, ...]], ...] = () keywords: tuple[tuple[TaintKind, tuple[str, ...]], ...] = () selects: bool = False + requires_called_result: TaintKind = TaintKind.NONE def kinds_at(self, position: int | None, keyword: str | None = None) -> TaintKind: """The kinds that must not reach the argument at ``position`` or passed as @@ -95,16 +99,30 @@ def kinds_at(self, position: int | None, keyword: str | None = None) -> TaintKin def merged(self, other: Sink) -> Sink: """This sink with ``other``'s kinds, positions and keywords added; it selects code - by name when either does, whatever the order the declarations merge in.""" + by name when either does, and a kind requires the called result only when every + declaration carrying it does, whatever the order the declarations merge in.""" + unconditional = (self.kinds & ~self.requires_called_result) | (other.kinds & ~other.requires_called_result) return Sink( self.symbol, self.kinds | other.kinds, self.positions + other.positions, self.keywords + other.keywords, self.selects or other.selects, + (self.requires_called_result | other.requires_called_result) & ~unconditional, ) + def applying(self, invoked: bool) -> Sink | None: + """This sink as it applies to one record of its call: to the call itself + (``invoked`` false), its unconditional kinds; to one invocation of the call's + result, the kinds that require it. ``None`` when no kind applies.""" + + conditional = self.kinds & self.requires_called_result + kinds = conditional if invoked else self.kinds & ~self.requires_called_result + if not kinds: + return None + return replace(self, kinds=kinds, requires_called_result=conditional & kinds) + @dataclass(frozen=True) class Sanitizer: diff --git a/tests/test_attribute_call_sinks.py b/tests/test_attribute_call_sinks.py new file mode 100644 index 0000000..8435ced --- /dev/null +++ b/tests/test_attribute_call_sinks.py @@ -0,0 +1,271 @@ +"""Acceptance tests for issue #240, part 2: ``getattr(obj, name)(...)`` with a name the +input chooses. + +The name only matters when the attribute it selects is called: the flow is anchored at +each invocation of ``getattr``'s result, one finding per invocation, in the function and +through project helpers, whose summaries keep the ``getattr`` call and each invocation. +What runs depends on what the object exposes: a hotspot, ``Untrusted attribute call``, +on a module too. A check proves the name only when it limits it to constants, and only +on the value ``getattr`` received: checking the name after it was rebound proves +nothing. Not followed, as documented: a ``getattr`` whose result is only read, returned, +or stored before it is called. +""" + +from __future__ import annotations + +import contextlib +import io +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.cli import main +from coretrace_python.findings import Finding +from coretrace_python.source import SourceManager + +HEADER = "import os\n\n\nclass Handler:\n def run(self):\n name = input()\n" + + +def injections(body: str) -> list[Finding]: + text = HEADER + "".join(f" {line}\n" for line in body.splitlines()) + findings = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return sorted((f for f in findings if f.rule_id == "code-injection"), key=lambda f: f.span.start_line) + + +def shape(finding: Finding) -> tuple[str, str, str, int, str | None, str | None]: + return ( + finding.metadata["verdict"], + finding.confidence.value, + finding.message.split(":", 1)[0], + finding.span.start_line, + finding.metadata.get("selected_line"), + finding.metadata.get("invoked_line"), + ) + + +ATTRIBUTE_CALL = ("hotspot", "medium", "Untrusted attribute call") + + +def test_the_called_attribute_is_reported_at_its_invocation() -> None: + (finding,) = injections("handler = getattr(self, name)\nhandler()") + + assert shape(finding) == (*ATTRIBUTE_CALL, 8, "7", "8") + + +def test_each_invocation_is_reported_once() -> None: + findings = injections("handler = getattr(self, name)\nhandler()\nhandler(1)") + + assert [shape(f) for f in findings] == [(*ATTRIBUTE_CALL, 8, "7", "8"), (*ATTRIBUTE_CALL, 9, "7", "9")] + + +def test_two_selections_reaching_one_invocation_are_reported_each() -> None: + body = "if name:\n handler = getattr(self, name)\nelse:\n handler = getattr(os, name)\nhandler()" + + findings = injections(body) + + assert sorted(f.metadata["selected_line"] for f in findings) == ["10", "8"] + assert {f.span.start_line for f in findings} == {11} + + +@pytest.mark.parametrize( + "body", + [ + # Only read, not called. + "value = getattr(self, name)", + "print(getattr(self, name, None))", + # Not followed (documented): returned, or stored before it is called. + "return getattr(self, name)", + "self.handler = getattr(self, name)\nself.handler()", + # The input is the object, not the name. + "getattr(name, 'strip')()", + "getattr(self, 'run')()", + ], +) +def test_a_name_that_selects_nothing_called_is_not_reported(body: str) -> None: + assert injections(body) == [] + + +@pytest.mark.parametrize( + "body", + [ + "getattr(os, name)('id')", + "getattr(self, 'do_' + name)()", + "handler = getattr(self, name)\nif callable(handler):\n handler()", + "if name.isidentifier():\n getattr(self, name)()", + ], +) +def test_a_module_a_prefix_or_a_check_of_the_shape_leaves_a_hotspot(body: str) -> None: + (finding,) = injections(body) + + assert shape(finding)[:3] == ATTRIBUTE_CALL + + +@pytest.mark.parametrize( + "body", + [ + "if name in ('start', 'stop'):\n getattr(self, name)()", + "handler = getattr(self, name)\nif name in ('start', 'stop'):\n handler()", + "handler = getattr(self, name)\nif name == 'start':\n handler()", + ], +) +def test_a_check_against_constants_between_selection_and_invocation_proves_the_name(body: str) -> None: + assert injections(body) == [] + + +@pytest.mark.parametrize( + "body", + [ + # The name checked is another value than the one ``getattr`` received. + "handler = getattr(self, name)\nname = 'start'\nif name in ('start', 'stop'):\n handler()", + "handler = getattr(self, name)\nname = input()\nif name in ('start', 'stop'):\n handler()", + "handler = getattr(self, name)\nchecked = name.strip()\nif checked in ('start', 'stop'):\n handler()", + ], +) +def test_a_check_on_another_value_than_the_selected_name_proves_nothing(body: str) -> None: + (finding,) = injections(body) + + assert shape(finding)[:3] == ATTRIBUTE_CALL + + +@pytest.mark.parametrize( + "body", + [ + "eval(name)()", + "result = eval(name)\nresult()\nresult(1)", + ], +) +def test_an_unconditional_sink_whose_result_is_called_is_reported_once(body: str) -> None: + (finding,) = injections(body) + + assert (finding.message.split(":", 1)[0], finding.span.start_line) == ("Code injection", 7) + + +HELPERS = { + "app/__init__.py": "", + "app/dispatch.py": ( + "def invoke(obj, name):\n" + " handler = getattr(obj, name)\n" + " handler()\n" + " handler(1)\n" + "\n\n" + "def forward(obj, name):\n" + " invoke(obj, name)\n" + "\n\n" + "def evaluate(code):\n" + " return eval(code)()\n" + ), + "app/views.py": ( + "from flask import request\n\nfrom app.dispatch import evaluate, forward, invoke\n\n\n" + "class View:\n" + " def direct(self):\n" + " invoke(self, request.args['a'])\n\n" + " def nested(self):\n" + " forward(self, request.args['a'])\n\n" + " def code(self):\n" + " evaluate(request.args['c'])\n" + ), +} + + +def project_findings(root: Path, **options: object) -> list[tuple[str, str, int, str | None, str | None]]: + analysis = engine.analyze_project(root, [engine.BUNDLED_PLUGINS], **options) # type: ignore[arg-type] + return sorted( + ( + f.function or "", + f.message.split(":", 1)[0], + f.span.start_line, + f.metadata.get("sink_line"), + f.metadata.get("invoked_line"), + ) + for f in analysis.findings + if f.rule_id == "code-injection" + ) + + +EXPECTED_THROUGH_HELPERS = [ + ("code", "Code injection", 14, "12", None), + ("direct", "Untrusted attribute call", 8, "2", "3"), + ("direct", "Untrusted attribute call", 8, "2", "4"), + ("nested", "Untrusted attribute call", 11, "2", "3"), + ("nested", "Untrusted attribute call", 11, "2", "4"), +] + + +def write(root: Path, files: dict[str, str]) -> Path: + for name, text in files.items(): + (root / name).parent.mkdir(parents=True, exist_ok=True) + (root / name).write_text(text, encoding="utf-8") + return root + + +def test_helper_summaries_keep_the_selection_and_each_invocation(tmp_path: Path) -> None: + root = write(tmp_path / "project", HELPERS) + + assert project_findings(root) == EXPECTED_THROUGH_HELPERS + + +@pytest.mark.parametrize("cached", [False, True], ids=["workers", "workers-and-warm-cache"]) +def test_workers_and_the_cache_keep_every_invocation(tmp_path: Path, cached: bool) -> None: + root = write(tmp_path / "project", HELPERS) + cache = ProjectCache(tmp_path / "cache") if cached else None + if cache is not None: + engine.analyze_project(root, [engine.BUNDLED_PLUGINS], cache=cache) + + assert project_findings(root, jobs=2, cache=cache) == EXPECTED_THROUGH_HELPERS + + +MANIFEST = ( + 'name = "getattr-sink"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = []\n\n[entrypoint]\nmodule = "getattr_sink"\nclass = "GetattrSink"\n' +) +# Another plugin declaring every ``getattr`` name a CODE sink, called or not. +PLUGIN = ''' +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import Sink, TaintKind + + +class GetattrSink(ModelPlugin): + name: ClassVar[str] = "getattr-sink" + models: ClassVar = ( + Sink(SymbolId("python.builtins.getattr"), TaintKind.CODE, ((TaintKind.CODE, (1,)),)), + ) +''' + + +@pytest.mark.parametrize("extra_first", [True, False], ids=["extra-plugin-first", "bundled-first"]) +def test_an_unconditional_declaration_reports_at_the_selection_without_duplicate(tmp_path: Path, extra_first: bool) -> None: + plugin = tmp_path / "plugins" / "getattr_sink" + plugin.mkdir(parents=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "getattr_sink.py").write_text(PLUGIN, encoding="utf-8") + root = write( + tmp_path / "project", + {"app.py": "from flask import request\n\n\nclass View:\n def run(self):\n handler = getattr(self, request.args['a'])\n handler()\n handler(1)\n"}, + ) + roots = [tmp_path / "plugins", engine.BUNDLED_PLUGINS] + + findings = engine.analyze_project(root, roots if extra_first else roots[::-1]).findings + + assert [(f.span.start_line, f.metadata.get("invoked_line")) for f in findings if f.rule_id == "code-injection"] == [ + (6, None) + ] + + +def test_the_command_line_reports_the_invocation(tmp_path: Path) -> None: + root = write(tmp_path / "project", {"app.py": HEADER + " getattr(self, name)()\n"}) + out = io.StringIO() + with contextlib.redirect_stdout(out): + main(["--check", str(root), "--format", "json"]) + + (finding,) = [f for f in json.loads(out.getvalue())["findings"] if f["rule_id"] == "code-injection"] + assert (finding["location"]["line"], finding["metadata"]["selected_line"], finding["metadata"]["invoked_line"]) == ( + 7, + "7", + "7", + ) diff --git a/tests/test_sink_called_result.py b/tests/test_sink_called_result.py new file mode 100644 index 0000000..baf37e0 --- /dev/null +++ b/tests/test_sink_called_result.py @@ -0,0 +1,66 @@ +"""The contract of ``Sink.requires_called_result`` (#240, part 2): the kinds of a sink that +apply only when the call's result is itself called. + +A kind requires the called result only when every declaration carrying it does: merging +never weakens an unconditional kind, whatever the order, and another kind added to the +same sink, such as an advisory's, does not make a conditional one unconditional. A +record of the call applies the unconditional kinds to the selection and the conditional +ones to each invocation of its result. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import ModelTable, Sink, TaintKind + +GETATTR = SymbolId("python.builtins.getattr") +CODE, ADVISORY, PATH = TaintKind.CODE, TaintKind.ADVISORY, TaintKind.PATH +NONE = TaintKind.NONE + + +def sink(kinds: TaintKind, requires: TaintKind = NONE) -> Sink: + return Sink(GETATTR, kinds, requires_called_result=requires) + + +@pytest.mark.parametrize( + ("first", "second", "kinds", "requires"), + [ + # Both conditional: conditional. + (sink(CODE, CODE), sink(CODE, CODE), CODE, CODE), + # One unconditional declaration of the kind: unconditional. + (sink(CODE, CODE), sink(CODE), CODE, NONE), + # Another kind, unconditional, leaves the conditional one as it was. + (sink(CODE, CODE), sink(ADVISORY), CODE | ADVISORY, CODE), + # Kinds of each side keep their own requirement. + (sink(CODE | PATH, CODE), sink(PATH, PATH), CODE | PATH, CODE), + ], +) +def test_a_kind_requires_the_called_result_only_when_every_declaration_does( + first: Sink, second: Sink, kinds: TaintKind, requires: TaintKind +) -> None: + for merged in (first.merged(second), second.merged(first)): + assert (merged.kinds, merged.requires_called_result) == (kinds, requires) + + +def test_an_advisory_sink_added_to_the_table_keeps_the_conditional_kind() -> None: + table = ModelTable((), (sink(CODE, CODE),), ()).extended(Sink(GETATTR, ADVISORY)) + merged = table.sink(GETATTR) + + assert merged is not None + assert merged.requires_called_result == CODE + + +def test_a_record_applies_the_kinds_its_record_calls_for() -> None: + both = Sink(GETATTR, CODE | ADVISORY, ((CODE, (1,)),), selects=True, requires_called_result=CODE) + + selection = both.applying(invoked=False) + invocation = both.applying(invoked=True) + + assert selection is not None and selection.kinds == ADVISORY + assert invocation is not None and invocation.kinds == CODE + # What else the sink says is kept. + assert invocation.selects and invocation.kinds_at(1) == CODE and invocation.kinds_at(0) == NONE + assert sink(CODE, CODE).applying(invoked=False) is None + assert sink(CODE).applying(invoked=True) is None From 00afd54c75814e6ffee505357b238505ea1b03fb Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 05:59:53 +0200 Subject: [PATCH 2/4] test(regression): record the getattr hotspot in luigi's RPC handler luigi/server.py:101 calls getattr(self._scheduler, method) with the route's method; RPC_METHODS is a module-level registry, so the guard proves nothing and the flow is a hotspot. --- tests/regression/expected/luigi.json | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/regression/expected/luigi.json b/tests/regression/expected/luigi.json index 16ecf57..de1045d 100644 --- a/tests/regression/expected/luigi.json +++ b/tests/regression/expected/luigi.json @@ -170,6 +170,13 @@ "rule": "unsafe-archive-extraction", "function": "safe_extract" }, + { + "file": "luigi/server.py", + "line": 101, + "rule": "code-injection", + "function": "get", + "verdict": "hotspot" + }, { "file": "test/_test_ftp.py", "line": 43, From 2f2c23ed903fd4cd5581fc9660738efa2d5a2b7a Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 06:39:46 +0200 Subject: [PATCH 3/4] fix(code-injection): judge both sites of an attribute call and resolve calls of call results A check proving the name where getattr selects holds past a join; a flow is refuted when its selection or its invocation cannot run, and a summary keeps an invocation only when both can. Calling the result of a call, or an item of it, now calls its __call__ instead of the symbol it comes from, so its arguments are not read as the original call's. Part of #240. --- docs/usage.md | 2 +- src/coretrace_python/findings/refutation.py | 27 ++++- .../interprocedural/callgraph.py | 15 ++- .../interprocedural/summaries.py | 34 ++++-- tests/test_attribute_call_review.py | 111 ++++++++++++++++++ 5 files changed, 175 insertions(+), 14 deletions(-) create mode 100644 tests/test_attribute_call_review.py diff --git a/docs/usage.md b/docs/usage.md index aac0310..c953d32 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -121,7 +121,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| | `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. The arguments after a script or code are the script's own: options of unknown meaning, reported `medium` (a hotspot after an option of the script), refuted behind a constant prefix. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. Every case keeps the rule id `command-injection`, so suppressions and baselines match whatever the case. When the rule reads the command element by element and establishes what the input is there, the finding's `injection` metadata (in the JSON report, and as a property of its SARIF result) says so: `command` when the input runs or chooses what runs (a shell string, the program, an operand run as a command, the value of an option that runs it); `option` when it can only reach the program as an option or as the value of an option not established to run it, and the message is then titled `Option injection`. When that cannot be established (an option whose effect is to review, an unknown program or element before the input, a command that is not a list or tuple display, a sink such as `os.system` the rule does not read element by element), the metadata is absent and the title stays `Command injection`. | -| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. Also the name `getattr(obj, name)` receives, when the attribute it selects is called: the finding, titled `Untrusted attribute call`, is placed at each call of the result, one per call, with the lines of the selection and of the call in its metadata (`selected_line`, `invoked_line`), and through project functions at the call that reaches them. What runs depends on what the object exposes, so it is a hotspot, on a module too and behind a constant prefix; a check proves the name only when it limits the value `getattr` received to constants, wherever it stands before the call, and a check of a name rebound since proves nothing. A `getattr` whose result is only read is not reported, nor, in this version, one whose result is returned to the caller, stored in an attribute or a container, or passed on before it is called, nor `operator.attrgetter`, `operator.methodcaller`, `obj.__getattribute__(name)` or `vars(obj)[name]`; checks made inside a project function are not seen from its callers, as for every sink. | +| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. Also the name `getattr(obj, name)` receives, when the attribute it selects is called: the finding, titled `Untrusted attribute call`, is placed at each call of the result, one per call, with the lines of the selection and of the call in its metadata (`selected_line`, `invoked_line`), and through project functions at the call that reaches them. What runs depends on what the object exposes, so it is a hotspot, on a module too and behind a constant prefix; a check proves the name only when it limits the value `getattr` received to constants, where `getattr` selects (even past a join) or before the call, and a check of a name rebound since proves nothing; nothing is reported when the selection or the call cannot run, in the function and inside a project function. Calling the result of a call, or an item of it, calls what the call returned: its arguments are not read as the call's, so `getattr(self, 'factory')(None, data)` or `eval('lambda x: x')(data)` reports nothing. A `getattr` whose result is only read is not reported, nor, in this version, one whose result is returned to the caller, stored in an attribute or a container, or passed on before it is called, nor `operator.attrgetter`, `operator.methodcaller`, `obj.__getattribute__(name)` or `vars(obj)[name]`; checks made inside a project function are not seen from its callers, as for every sink. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | | `path-traversal` | A file system path (`open`, `send_file`, `os.remove`, …). | diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index 5cc44ab..c8f174f 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -211,9 +211,20 @@ def covered(self, origin: Value, argument: Value, proven: Mapping[Value, str]) - # ------------------------------------------------------------------ guards def sink_block(self, flow: TaintFlow) -> BlockId | None: + return self.call_block(flow.location) + + def selection_block(self, flow: TaintFlow) -> BlockId | None: + """The block of the call that selected what an invocation runs, when both are in + this function (``getattr``'s, #240); ``None`` for any other flow.""" + + if flow.invoked_at is None or flow.through is not None or flow.sink_location is None: + return None + return self.call_block(flow.sink_location) + + def call_block(self, location: SourceSpan) -> BlockId | None: for block in self.function.blocks: for instruction in block.instructions: - if isinstance(instruction, Call) and instruction.location == flow.location: + if isinstance(instruction, Call) and instruction.location == location: return block.id return None @@ -469,9 +480,14 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: chain = self.closure(flow.argument) | {flow.argument} # A sink that selects code by name is proven only by constants (``Sink.selects``). selects = flow.sink.selects + # A flow anchored at an invocation is judged at its selection too: a check proving + # the selected name there proves it for every invocation, past a join (#240). + selection = self.selection_block(flow) + sites = (sink,) if selection is None else (selection, sink) proven: dict[Value, str] = { value: f"numeric value within {interval}" - for value, interval in self.ranges.at(sink).items() + for site in sites + for value, interval in self.ranges.at(site).items() if value in chain and not selects } mentions: dict[Value, int] = {} @@ -480,7 +496,7 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: if self.authorization is not None else None ) - for guard in self.guards(sink): + for guard in dict.fromkeys(guard for site in sites for guard in self.guards(site)): found, mentioned = self.interpret(guard.condition, guard.truth) for value, (reason, kinds, exact) in found.items(): # A check proving some kinds only leaves the others reaching the sink. @@ -533,7 +549,10 @@ def judge_flows( verdicts = [] for flow in taint.flows: sink = judge.sink_block(flow) - verdicts.append(judge.judge(flow, sink is not None and sink in reachable)) + selection = judge.selection_block(flow) + # Both sites of a flow anchored at an invocation must be able to run (#240). + runs = sink is not None and sink in reachable and (selection is None or selection in reachable) + verdicts.append(judge.judge(flow, runs)) return Verdicts(tuple(verdicts)) diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index de087c0..e52ce39 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -270,6 +270,19 @@ def derive_symbols( return symbols +def called_symbol(symbol: SymbolId, made: Instruction | None, table: SymbolTable) -> SymbolId: + """What calling a value that denotes ``symbol`` calls. A symbol, or an attribute of + one, is the callable itself. The result of a call (entered with ``with`` or awaited) + and an item denote where they come from, so that their attributes resolve + (``tarfile.open(p).extractall``, ``request.files['f'].save``), but calling one calls + what the call returned or the container held, not the symbol it comes from: its + ``__call__``, so its arguments are never read as the original call's (#240).""" + + if isinstance(made, Call | WithEnter | Await | GetItem): + return table.attribute(symbol, "__call__") + return symbol + + # Conditions name short constants (``True``, ``None``, a mode); a log message or a query # passed as an argument is not recorded, so call sites stay small in the cache. MAX_CONSTANT_LENGTH = 64 @@ -504,7 +517,7 @@ def resolve_targets( inherited[instruction.result] = table.attribute(base, instruction.attribute) symbols = derive_symbols(function, table, inherited) targets: dict[Value, Target] = { - value: ExternalSymbol(symbol) for value, symbol in symbols.items() + value: ExternalSymbol(called_symbol(symbol, defs.get(value), table)) for value, symbol in symbols.items() } for block in function.blocks: for instruction in block.instructions: diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 5e807fe..028dd1c 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -15,7 +15,7 @@ from __future__ import annotations import dataclasses -from collections.abc import Iterable, Iterator, Mapping +from collections.abc import Callable, Iterable, Iterator, Mapping from dataclasses import dataclass from functools import cached_property from types import MappingProxyType @@ -24,6 +24,7 @@ from coretrace_python.abstract import ( ATTRIBUTES, ELEMENTS, + ConstantPropagation, HeapAnalysis, HeapFacts, HeapLocation, @@ -418,9 +419,11 @@ def __init__( heap: HeapFacts | None = None, clearing: Clearing | None = None, templates: TemplateCalls | None = None, + reachable: Callable[[BlockId], bool] | None = None, ) -> None: self.name = name self.function = function + self.reachable = reachable or (lambda block: True) self.graph = graph self.table = table self.project = project or {} @@ -431,6 +434,9 @@ def __init__( self.defs: dict[Value, Instruction] = { i.result: i for block in function.blocks for i in block.instructions if i.result } + self.block_of: dict[int, BlockId] = { + id(i): block.id for block in function.blocks for i in block.instructions + } self.external: dict[_CallKey, ExternalCall] = {} self.returns: Dep = EMPTY self.stores: dict[HeapLocation, Dep] = {} @@ -570,10 +576,12 @@ def call(self, call: Call, state: dict[Key, Dep]) -> Dep: given = self.graph.arguments_at(self.name, call.location) self.record(target.symbol, arguments, named, call.location, None, given) # Where its result is called: facts the summary keeps whatever the models; - # the consumer decides which kinds of a sink apply there (#240). - if call.result is not None: + # the consumer decides which kinds of a sink apply there (#240). An invocation + # is kept only when it and the call can run, by constant propagation. + if call.result is not None and self.runs(call): for invocation in invocations(self.function, self.uses, call.result): - self.record(target.symbol, arguments, named, call.location, None, given, invocation.location) + if self.runs(invocation): + self.record(target.symbol, arguments, named, call.location, None, given, invocation.location) self.record_filters(target.symbol, call, given, state) cleared = everything.clearing(self.clearing.clears(target.symbol, given)) return cleared | Dep(externals=frozenset({target.symbol})) @@ -673,6 +681,11 @@ def mapped(indices: Dependencies, cleared: Cleared = ()) -> Dep: def uses(self) -> DefUse: return def_use(self.function) + def runs(self, call: Call) -> bool: + """Whether ``call`` can run: its block is reachable by constant propagation.""" + + return self.reachable(self.block_of[id(call)]) + def record_filters(self, symbol: SymbolId, call: Call, given: Arguments, state: Mapping[Key, Dep]) -> None: """The filter calls a render makes, each at the filter in its template, with what the context entries reaching its value and its argument depend on. The request the @@ -746,8 +759,9 @@ def summarize( static: bool = False, clearing: Clearing | None = None, templates: TemplateCalls | None = None, + reachable: Callable[[BlockId], bool] | None = None, ) -> FunctionSummary: - problem = _DependenceProblem(name, function, graph, table, project, heap, clearing, templates) + problem = _DependenceProblem(name, function, graph, table, project, heap, clearing, templates, reachable) solution = solve(problem, cfg) problem.external = {} problem.returns = EMPTY @@ -790,6 +804,7 @@ class SummaryAnalysis(Analysis[SummaryTable]): CallGraphAnalysis, SSAAnalysis, CFGAnalysis, + ConstantPropagation, ProjectSummaries, HeapAnalysis, ClearingAnalysis, @@ -804,7 +819,7 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: clearing = ctx.get(ClearingAnalysis) templates = ctx.get(TemplateCallsAnalysis) table: dict[str, FunctionSummary] = {} - supported: dict[str, tuple[FunctionIR, CFG, HeapFacts]] = {} + supported: dict[str, tuple[FunctionIR, CFG, HeapFacts, Callable[[BlockId], bool]]] = {} for name, function in graph.definitions.items(): if name in graph.unsupported: count = len(function.parameters) @@ -814,14 +829,17 @@ def compute(cls, ctx: AnalysisContext) -> SummaryTable: ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function), ctx.get(HeapAnalysis, function), + ctx.get(ConstantPropagation, function).reachable, ) table[name] = FunctionSummary(name, len(function.parameters), NONE, ()) changed = True while changed: changed = False - for name, (ssa, cfg, heap) in supported.items(): + for name, (ssa, cfg, heap, reachable) in supported.items(): static = _is_staticmethod(graph.definitions[name]) - updated = summarize(name, ssa, cfg, graph, table, project, heap, static, clearing, templates) + updated = summarize( + name, ssa, cfg, graph, table, project, heap, static, clearing, templates, reachable + ) if updated != table[name]: table[name] = updated changed = True diff --git a/tests/test_attribute_call_review.py b/tests/test_attribute_call_review.py new file mode 100644 index 0000000..eb76748 --- /dev/null +++ b/tests/test_attribute_call_review.py @@ -0,0 +1,111 @@ +"""Review of #240, part 2 (PR #257): what anchoring a flow at the invocation must keep. + +The selection and the invocation are two sites of one flow. A check that proves the name +where ``getattr`` selects proves it for every invocation, even past a join, and a check +before the invocation counts too; the flow is refuted when either site cannot run, in the +function and, for the invocation, through a project helper. Calling the result of a call, +or an item of it, does not call the symbol it comes from: its arguments are not the +call's, so ``h(None, input())`` with ``h = getattr(self, 'factory')`` selects nothing, and +``eval('lambda x: x')(input())`` evaluates constant code. The chains that resolve the +attributes of a result are kept. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.source import SourceManager + + +def findings(text: str, rule: str) -> list[Finding]: + analysed = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return [f for f in analysed if f.rule_id == rule] + + +def method(body: str) -> str: + return "import os\n\n\nclass Handler:\n def run(self):\n name = input()\n" + "".join( + f" {line}\n" for line in body.splitlines() + ) + + +@pytest.mark.parametrize( + "guard", + ["name in ('start', 'stop')", "name == 'start'"], +) +def test_a_check_proving_the_name_at_the_selection_holds_past_a_join(guard: str) -> None: + body = f"if {guard}:\n handler = getattr(self, name)\nelse:\n handler = print\nhandler()" + + assert findings(method(body), "code-injection") == [] + + +def test_a_check_of_the_shape_at_the_selection_still_leaves_a_hotspot() -> None: + body = "if name.isidentifier():\n handler = getattr(self, name)\nelse:\n handler = print\nhandler()" + + (finding,) = findings(method(body), "code-injection") + assert finding.metadata["verdict"] == "hotspot" + + +@pytest.mark.parametrize( + "body", + [ + # The selection cannot run. + "if False:\n handler = getattr(self, name)\nelse:\n handler = print\nhandler()", + # The invocation cannot run. + "handler = getattr(self, name)\nif False:\n handler()", + ], +) +def test_a_selection_or_an_invocation_that_cannot_run_is_not_reported(body: str) -> None: + assert findings(method(body), "code-injection") == [] + + +HELPER = "def select(obj, name):\n{body}\n\n\nclass Handler:\n def run(self):\n select(self, input())\n" + + +@pytest.mark.parametrize( + ("body", "reported"), + [ + (" handler = getattr(obj, name)\n if False:\n handler()", False), + (" if False:\n getattr(obj, name)()", False), + (" handler = getattr(obj, name)\n handler()", True), + ], +) +def test_through_a_helper_both_sites_must_be_able_to_run(body: str, reported: bool) -> None: + text = HELPER.format(body=body) + found = findings(text, "code-injection") + + # Reported at the call of the helper, the last line. + assert [f.span.start_line for f in found] == ([text.count("\n")] if reported else []) + + +FACTORY = "class Handler:\n def factory(self, unused, text):\n return print\n\n def run(self):\n" + + +@pytest.mark.parametrize( + "text", + [ + # The review's case: the input is an argument of the selected method. + FACTORY + " h = getattr(self, 'factory')\n h(None, input())()\n", + FACTORY + " getattr(self, 'factory')(None, input())()\n", + # The same confusion with the other sinks of the rule. + "def run():\n eval('lambda x: x')(input())\n", + "import importlib\n\n\ndef run():\n importlib.import_module('plugins')(input())\n", + "import runpy\n\n\ndef run():\n runpy.run_path('jobs/report.py')['main'](input())\n", + ], +) +def test_the_arguments_of_a_returned_callable_are_not_the_call_s(text: str) -> None: + assert findings(text, "code-injection") == [] + + +@pytest.mark.parametrize( + ("text", "rule", "line"), + [ + ("def run():\n eval(input())('x')\n", "code-injection", 2), + ("class Handler:\n def run(self):\n getattr(self, input())(None)()\n", "code-injection", 3), + ("import runpy\n\n\ndef run():\n runpy.run_path(input()).get('main')\n", "code-injection", 5), + ("import tarfile\n\n\ndef run(p):\n tarfile.open(p).extractall('/tmp/x')\n", "unsafe-archive-extraction", 5), + ], +) +def test_the_call_itself_and_the_chains_on_its_result_are_still_reported(text: str, rule: str, line: int) -> None: + assert [f.span.start_line for f in findings(text, rule)] == [line] From ae6bdb95155d63dfeac4b36a38436df303534734 Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 15 Oct 2026 07:54:34 +0200 Subject: [PATCH 4/4] fix(refutation): keep a check at the invocation from proving a name selected in an earlier iteration When the callable reaches its invocation across a loop's back edge, it was selected with an earlier instance of the name than the one a check at the invocation tests: only the checks where getattr selects prove it there. Part of #240. --- docs/usage.md | 2 +- src/coretrace_python/findings/refutation.py | 54 +++++++++++- tests/test_attribute_call_loops.py | 91 +++++++++++++++++++++ 3 files changed, 142 insertions(+), 5 deletions(-) create mode 100644 tests/test_attribute_call_loops.py diff --git a/docs/usage.md b/docs/usage.md index c953d32..94c739f 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -121,7 +121,7 @@ variable is not treated that way yet. | Rule | Reached sink | |---|---| | `command-injection` | Shell or process execution (`os.system`, `subprocess` with a string, …), judged on what the input controls in the command. A string a shell runs, a string naming the program, the program itself, or an operand the program runs as a command (after the host for `ssh`, the program of `sudo`, `env`, `xargs`, `timeout`…, any argument of a batch file) is `high`. The script an interpreter runs (`bash run.sh`, `python tool.py`, its first operand when no option such as `-c` gives the code), or the module of `python -m`, is `high` and kept for review: execution is possible, and its exploitability depends on the file the input chooses. The arguments after a script or code are the script's own: options of unknown meaning, reported `medium` (a hotspot after an option of the script), refuted behind a constant prefix. Otherwise an element of an argument list run without a shell can only be read as an option: `medium`, or `high` when a documented option of the program runs a command it chooses (`git push --receive-pack=…`, `ssh -o ProxyCommand=…`). A value a documented option takes is judged by what the option does with it, whatever its prefix: run it (`sh -c`, `cmd /c`, `python -c`: `high`), keep it as data (`git commit -m`: refuted). After an option the rule does not know, the finding stays a hotspot to review. A constant prefix that keeps a standalone input from being an option (`f"v{version}"`), or an operand after `--` for a program that honours it, is refuted. Every case keeps the rule id `command-injection`, so suppressions and baselines match whatever the case. When the rule reads the command element by element and establishes what the input is there, the finding's `injection` metadata (in the JSON report, and as a property of its SARIF result) says so: `command` when the input runs or chooses what runs (a shell string, the program, an operand run as a command, the value of an option that runs it); `option` when it can only reach the program as an option or as the value of an option not established to run it, and the message is then titled `Option injection`. When that cannot be established (an option whose effect is to review, an unknown program or element before the input, a command that is not a list or tuple display, a sink such as `os.system` the rule does not read element by element), the metadata is absent and the title stays `Command injection`. | -| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. Also the name `getattr(obj, name)` receives, when the attribute it selects is called: the finding, titled `Untrusted attribute call`, is placed at each call of the result, one per call, with the lines of the selection and of the call in its metadata (`selected_line`, `invoked_line`), and through project functions at the call that reaches them. What runs depends on what the object exposes, so it is a hotspot, on a module too and behind a constant prefix; a check proves the name only when it limits the value `getattr` received to constants, where `getattr` selects (even past a join) or before the call, and a check of a name rebound since proves nothing; nothing is reported when the selection or the call cannot run, in the function and inside a project function. Calling the result of a call, or an item of it, calls what the call returned: its arguments are not read as the call's, so `getattr(self, 'factory')(None, data)` or `eval('lambda x: x')(data)` reports nothing. A `getattr` whose result is only read is not reported, nor, in this version, one whose result is returned to the caller, stored in an attribute or a container, or passed on before it is called, nor `operator.attrgetter`, `operator.methodcaller`, `obj.__getattribute__(name)` or `vars(obj)[name]`; checks made inside a project function are not seen from its callers, as for every sink. | +| `code-injection` | The code `eval` or `exec` runs; the globals and locals passed with it are data. `dangerous-eval` reports every such call, this rule the ones attacker input reaches. Also the module `importlib.import_module`, `__import__` or `runpy.run_module` imports, and the file `runpy.run_path` runs, by position or by keyword (`name`, `mod_name`, `path_name`); the package of a relative name, the globals and the run name are data. An import runs the top-level code of a module the input chooses among those that can be found, a script path the file it chooses: execution is possible, and how far it goes depends on what the input can reach, so these findings are titled `Untrusted module import` and `Untrusted script execution`. A constant package prefix (`"app.plugins." + name`) confines an import to that package: a hotspot. The environment and the command line reach these sinks as they reach `eval`. Also the directory `sys.path.insert` (its second argument), `sys.path.append` and `sys.path.extend` add to the import path, and the directory `site.addsitedir` adds (`sitedir`), judged the same way. A directory on the import path runs code only when a later import resolves to a module the attacker placed there: a hotspot, titled `Untrusted import path`. `site.addsitedir` runs the import lines of the directory's `.pth` files during the call: a vulnerability, titled `Untrusted site directory`, exploitable when the directory holds files the attacker controls, as the finding's evidence states. Coverage of the import path is partial: writes to `sys.path` that are not calls, a slice assignment (`sys.path[:0] = [d]`), `sys.path += [d]` or rebinding `sys.path`, are not reported. Also the name `getattr(obj, name)` receives, when the attribute it selects is called: the finding, titled `Untrusted attribute call`, is placed at each call of the result, one per call, with the lines of the selection and of the call in its metadata (`selected_line`, `invoked_line`), and through project functions at the call that reaches them. What runs depends on what the object exposes, so it is a hotspot, on a module too and behind a constant prefix; a check proves the name only when it limits the value `getattr` received to constants, where `getattr` selects (even past a join) or before the call, and a check of a name rebound since proves nothing; a check before the call does not prove a callable selected in an earlier iteration of a loop, which reaches the call across the loop's back edge: only a check where it was selected does (even when the name never changes, which this version does not establish); nothing is reported when the selection or the call cannot run, in the function and inside a project function. Calling the result of a call, or an item of it, calls what the call returned: its arguments are not read as the call's, so `getattr(self, 'factory')(None, data)` or `eval('lambda x: x')(data)` reports nothing. A `getattr` whose result is only read is not reported, nor, in this version, one whose result is returned to the caller, stored in an attribute or a container, or passed on before it is called, nor `operator.attrgetter`, `operator.methodcaller`, `obj.__getattribute__(name)` or `vars(obj)[name]`; checks made inside a project function are not seen from its callers, as for every sink. | | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | | `path-traversal` | A file system path (`open`, `send_file`, `os.remove`, …). | diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index c8f174f..71cf3dc 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -56,6 +56,7 @@ GetIter, Global, Instruction, + Phi, Symbol, UnaryOp, Value, @@ -154,6 +155,9 @@ def __init__( self.defs: dict[Value, Instruction] = { i.result: i for block in function.blocks for i in block.instructions if i.result } + self.defined_in: dict[Value, BlockId] = { + i.result: block.id for block in function.blocks for i in block.instructions if i.result + } self._closures: dict[Value, frozenset[Value]] = {} self._uses: dict[Value, list[Instruction]] | None = None @@ -222,12 +226,46 @@ def selection_block(self, flow: TaintFlow) -> BlockId | None: return self.call_block(flow.sink_location) def call_block(self, location: SourceSpan) -> BlockId | None: + found = self.call_at(location) + return None if found is None else found[0] + + def call_at(self, location: SourceSpan) -> tuple[BlockId, Call] | None: for block in self.function.blocks: for instruction in block.instructions: if isinstance(instruction, Call) and instruction.location == location: - return block.id + return block.id, instruction return None + def carried_across_iterations(self, flow: TaintFlow) -> bool: + """Whether the callable an invocation runs reaches it from its selection across a + loop's back edge, through the phis that merge it: selected in an earlier iteration, + with an earlier instance of the name than the one a check at the invocation tests + (#240).""" + + invocation = self.call_at(flow.location) + selection = None if flow.sink_location is None else self.call_at(flow.sink_location) + if invocation is None or selection is None or selection[1].result is None: + return False + selected = selection[1].result + pending: list[tuple[Value, bool]] = [(invocation[1].callee, False)] + seen: set[tuple[Value, bool]] = set() + while pending: + value, crossed = pending.pop() + if (value, crossed) in seen: + continue + seen.add((value, crossed)) + if value == selected: + if crossed: + return True + continue + made = self.defs.get(value) + if isinstance(made, Phi): + header = self.defined_in[value] + for incoming, predecessor in made.incoming: + # An edge from a block the phi's block dominates closes a loop. + pending.append((incoming, crossed or self.tree.dominates(header, predecessor))) + return False + def reaches(self, start: BlockId, target: BlockId, avoiding: BlockId) -> bool: seen: set[BlockId] = set() pending = [start] @@ -484,9 +522,13 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: # the selected name there proves it for every invocation, past a join (#240). selection = self.selection_block(flow) sites = (sink,) if selection is None else (selection, sink) + # The sites whose checks prove the selected name: both, unless the callable + # reaches its invocation across a back edge, where a check tests the name of + # another iteration than the one that selected it. + proving = (selection,) if selection is not None and self.carried_across_iterations(flow) else sites proven: dict[Value, str] = { value: f"numeric value within {interval}" - for site in sites + for site in proving for value, interval in self.ranges.at(site).items() if value in chain and not selects } @@ -496,11 +538,15 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: if self.authorization is not None else None ) - for guard in dict.fromkeys(guard for site in sites for guard in self.guards(site)): + guards: dict[_Guard, bool] = {} + for site in sites: + for guard in self.guards(site): + guards[guard] = guards.get(guard, False) or site in proving + for guard, proves in guards.items(): found, mentioned = self.interpret(guard.condition, guard.truth) for value, (reason, kinds, exact) in found.items(): # A check proving some kinds only leaves the others reaching the sink. - if value in chain and not flow.kinds & ~kinds and (exact or not selects): + if proves and value in chain and not flow.kinds & ~kinds and (exact or not selects): proven.setdefault(value, reason) examined = self.examined(guard.condition, mentioned, flow.argument, chain) for origin in origins: diff --git a/tests/test_attribute_call_loops.py b/tests/test_attribute_call_loops.py new file mode 100644 index 0000000..f32e55b --- /dev/null +++ b/tests/test_attribute_call_loops.py @@ -0,0 +1,91 @@ +"""Review of #240, part 2 (PR #257): a check before the invocation proves the selected +name only for the selection it can see. + +When the callable reaches its invocation across a loop's back edge, it was selected in +an earlier iteration: a check at the invocation tests the name of the current iteration, +not the one that selected it, so it proves nothing and the hotspot stays. A check where +``getattr`` selects still proves the name, and within one iteration a check before the +invocation does too. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.source import SourceManager + + +def injections(body: str) -> list[Finding]: + text = "class Handler:\n def run(self):\n" + "".join(f" {line}\n" for line in body.splitlines()) + findings = engine.check(SourceManager().add_source("app.py", text), [engine.BUNDLED_PLUGINS]) + return [f for f in findings if f.rule_id == "code-injection"] + + +def test_a_check_at_the_invocation_does_not_prove_a_name_selected_in_an_earlier_iteration() -> None: + # The review's case: with the inputs "dangerous", "start", "stop", the second + # iteration calls self.dangerous(), selected by the first. + body = ( + "name = input()\n" + "handler = print\n" + "for _ in range(2):\n" + " if name in ('start', 'stop'):\n" + " handler()\n" + " handler = getattr(self, name)\n" + " name = input()\n" + ) + + (finding,) = injections(body) + + assert (finding.metadata["verdict"], finding.span.start_line) == ("hotspot", 7) + assert (finding.metadata["selected_line"], finding.metadata["invoked_line"]) == ("8", "7") + + +def test_a_check_at_the_selection_proves_the_name_across_iterations() -> None: + body = ( + "name = input()\n" + "handler = print\n" + "for _ in range(2):\n" + " handler()\n" + " if name in ('start', 'stop'):\n" + " handler = getattr(self, name)\n" + " name = input()\n" + ) + + assert injections(body) == [] + + +@pytest.mark.parametrize( + "body", + [ + # Selected and called in the same iteration, the check in between. + ( + "for _ in range(2):\n name = input()\n handler = getattr(self, name)\n" + " if name in ('start', 'stop'):\n handler()\n" + ), + ( + "for _ in range(2):\n name = input()\n if name in ('start', 'stop'):\n" + " getattr(self, name)()\n" + ), + ], +) +def test_within_one_iteration_a_check_before_the_invocation_proves_the_name(body: str) -> None: + assert injections(body) == [] + + +def test_a_name_fixed_before_the_loop_is_not_proven_across_iterations_in_this_version() -> None: + # Sound to prove, since the name never changes, but the check is at the invocation + # and the callable crosses the back edge: the hotspot is kept. + body = ( + "name = input()\n" + "handler = print\n" + "for _ in range(2):\n" + " if name in ('start', 'stop'):\n" + " handler()\n" + " handler = getattr(self, name)\n" + ) + + (finding,) = injections(body) + + assert finding.metadata["verdict"] == "hotspot"