From eb15f33fd5dd502cbbd9ac2eae48e4a5c16e79be Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 13:10:59 +0200 Subject: [PATCH 1/2] test: define factory instances, loop else and match --- tests/test_factories_match_loop_else.py | 212 ++++++++++++++++++++++++ tests/test_hir_builder.py | 6 +- 2 files changed, 215 insertions(+), 3 deletions(-) create mode 100644 tests/test_factories_match_loop_else.py diff --git a/tests/test_factories_match_loop_else.py b/tests/test_factories_match_loop_else.py new file mode 100644 index 0000000..52c0daf --- /dev/null +++ b/tests/test_factories_match_loop_else.py @@ -0,0 +1,212 @@ +"""Acceptance tests for the last gaps found on the ``tests-project/`` repositories. + +- Factory instances: ``app = create_app()`` at module level, where ``create_app`` is a + project function whose summary returns a ``Flask`` instance, makes ``app.route`` an + entry point like ``app = Flask(__name__)`` does. One repository's routes were invisible. +- Loop ``else`` clauses run when the loop is exhausted without ``break``. +- ``match`` statements over literal patterns, wildcards, captures, or-patterns and + guards lower to an ``if`` chain over a hidden subject; other patterns are reported. + +Expected to remain red until ``While.orelse`` / ``For.orelse``, ``Match`` desugaring and +factory instances 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.ir.lowering import lower_module +from coretrace_python.ir.printer import format_module +from coretrace_python.source import SourceManager + +try: + from coretrace_python.taint.engine import factory_instances +except ImportError as error: # pragma: no cover - red until factories land + MISSING: Exception | None = error +else: + MISSING = None + if "orelse" not in nodes.For.__dataclass_fields__: + MISSING = AttributeError("loops have no orelse") + + +@pytest.fixture(autouse=True) +def require_pass() -> None: + if MISSING is not None: + pytest.fail(f"factories, match and loop else are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def check(text: str, name: str = "app.py") -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source(name, text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, int]]: + return sorted((f.rule_id, f.span.start_line) for f in findings) + + +def project(root: Path, files: dict[str, str]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +def hir(text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("s.py", text)) + + +def printed(text: str) -> str: + return format_module(lower_module(hir(text))) + + +# --------------------------------------------------------------------------- factory instances + + +FACTORY = ( + "import os\nfrom flask import Flask, request, send_file\n\n" + "def create_app():\n _app = Flask(__name__)\n _app.config['X'] = 1\n return _app\n\n" + "app = create_app()\n\n" + "@app.route('/img/')\n" + "def image(name):\n" + " return send_file(os.path.join('/srv/img', name + '.jpg'))\n" +) + + +def test_routes_of_a_factory_made_app_are_entry_points() -> None: + assert rules(check(FACTORY)) == [("path-traversal", 13)] + + +def test_factory_instances_are_derived_from_return_externals() -> None: + from coretrace_python.interprocedural import ProjectSummaries, SummaryAnalysis + from coretrace_python.semantic.scopes import ScopeAnalysis + from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId + + manager = engine.build_manager(hir(FACTORY)) + instances = factory_instances( + manager.module, + manager.get(ScopeAnalysis), + manager.get(SymbolAnalysis), + manager.get(SummaryAnalysis), + manager.get(ProjectSummaries), + ) + + assert instances == {"app": (SymbolId("python.flask.Flask"),)} + + +def test_factories_defined_in_another_file_work_through_the_project_index(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "factory.py": "from flask import Flask\n\ndef create_app():\n return Flask(__name__)\n", + "app.py": ( + "import os\nfrom flask import send_file\nfrom factory import create_app\n\n" + "app = create_app()\n\n" + "@app.route('/img/')\n" + "def image(name):\n" + " return send_file(os.path.join('/srv/img', name))\n" + ), + }, + ) + findings = engine.analyze_project(root, [PLUGINS]).findings + assert [(f.rule_id, Path(str(f.span.source_id)).name, f.span.start_line) for f in findings] == [ + ("path-traversal", "app.py", 9) + ] + + +def test_factories_returning_nothing_known_create_no_entry_point() -> None: + text = ( + "import os\n\ndef make():\n return object()\n\napp = make()\n\n" + "@app.route('/x/')\ndef image(name):\n os.system(name)\n" + ) + assert check(text) == () + + +# --------------------------------------------------------------------------- loop else + + +def test_loop_else_runs_only_when_the_loop_is_exhausted() -> None: + module = hir("def f(items):\n for item in items:\n if item:\n break\n else:\n return 'none'\n return item\n") + loop = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(loop, nodes.For) and len(loop.orelse) == 1 + + text = printed("def f(items):\n for item in items:\n if item:\n break\n else:\n return 'none'\n return item\n") + assert "for_next" in text and text.count("return") >= 2 + + module = hir("def f(n):\n while n:\n n = n - 1\n else:\n n = 0\n return n\n") + loop = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(loop, nodes.While) and len(loop.orelse) == 1 + + +def test_taint_reaches_the_else_clause_and_break_skips_it() -> None: + findings = check( + "import os\n\ndef f(items):\n for item in items:\n if item == 'stop':\n break\n" + " else:\n os.system(input())\n return 1\n" + ) + assert rules(findings) == [("command-injection", 8)] + + +# --------------------------------------------------------------------------- match + + +def test_match_over_literals_lowers_to_branches() -> None: + text = printed( + "def f(op, x):\n" + " match op:\n" + " case 'gray' | 'grey':\n return 1\n" + " case 'png' if x:\n return 2\n" + " case None:\n return 3\n" + " case other:\n return other\n" + ) + assert text.count("branch") >= 4 + assert "compare.eq" in text + + +def test_match_binds_captures_and_flows_taint() -> None: + findings = check( + "import os\n\ndef f():\n match input():\n case 'ls':\n os.system('ls')\n" + " case cmd:\n os.system(cmd)\n" + ) + assert rules(findings) == [("command-injection", 8)] + + +def test_match_wildcard_and_subject_taint() -> None: + findings = check( + "import os\n\ndef f(op):\n cmd = input()\n match op:\n case 'run':\n os.system(cmd)\n" + " case _:\n pass\n" + ) + assert rules(findings) == [("command-injection", 7)] + + +def test_unsupported_patterns_are_reported_per_function() -> None: + findings = check("def f(p):\n match p:\n case [x, y]:\n return x\n case _:\n return 0\n") + assert [f.rule_id for f in findings] == ["unsupported-syntax"] + assert "pattern" in findings[0].message + + +def test_the_image_editing_site_is_analysed(tmp_path: Path) -> None: + text = ( + "import os\nfrom flask import Flask, request\nfrom werkzeug.utils import secure_filename\n\n" + "app = Flask(__name__)\n\n" + "def process(filename, operation):\n" + " match operation:\n" + " case 'cgray':\n return filename + '.gray'\n" + " case 'cpng':\n return filename + '.png'\n" + " return filename\n\n" + "@app.route('/edit', methods=['POST'])\n" + "def edit():\n" + " f = request.files['file']\n" + " name = secure_filename(f.filename)\n" + " f.save(os.path.join('/srv', name))\n" + " return process(name, request.form.get('operation'))\n" + ) + assert check(text) == () diff --git a/tests/test_hir_builder.py b/tests/test_hir_builder.py index b963f57..8e0ed77 100644 --- a/tests/test_hir_builder.py +++ b/tests/test_hir_builder.py @@ -209,10 +209,10 @@ def test_represents_raise_with_and_without_exception() -> None: @pytest.mark.parametrize( "source_text, message", [ - ("while x:\n pass\nelse:\n pass\n", "else"), - ("for a, b in items:\n pass\nelse:\n pass\n", "else"), + ("def f(x):\n yield from x\n", "yield from"), + ("a, *rest = items\n", "starred"), ], - ids=["while-else", "for-else"], + ids=["yield-from", "starred-target"], ) def test_unsupported_control_flow_forms_are_reported(source_text: str, message: str) -> None: from coretrace_python.frontend import HIRBuildError From 3c9c1e89c4d0dc8462af56702356db40c9df4c22 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 13:10:59 +0200 Subject: [PATCH 2/2] feat: resolve factory-made instances, lay out loop else and match statements --- README.md | 10 ++- src/coretrace_python/cfg/builder.py | 53 +++++++++++++- src/coretrace_python/frontend/ast_adapter.py | 33 +++++++-- src/coretrace_python/hir/nodes.py | 69 ++++++++++++++++++ src/coretrace_python/semantic/scopes.py | 24 +++++-- src/coretrace_python/taint/engine.py | 75 ++++++++++++++++++-- 6 files changed, 244 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 05c5186..9938820 100644 --- a/README.md +++ b/README.md @@ -98,7 +98,9 @@ repeated. The repository ships a first set under `plugins/`: security models for standard library, Flask, FastAPI, Django, SQLAlchemy and the Requests and httpx clients (`models/`), taint detectors for SQL injection, command injection, path traversal, SSRF and XSS (`security/`), and syntactic detectors for `eval`/`exec` and weak hashes (`syntax/`). Route handlers of Flask and -FastAPI receive their parameters as HTTP input; `request.args` and its siblings are HTTP +FastAPI receive their parameters as HTTP input, whether the application is created +directly (`app = Flask(__name__)`) or by a project factory whose summary returns it +(`app = create_app()`); `request.args` and its siblings are HTTP sources. Django views are undecorated, so a parameter annotated with a request class (`request: HttpRequest`), a method of a class-based view and a view decorated by one of Django's or Django REST framework's view decorators receive HTTP input; a bare @@ -145,8 +147,10 @@ with name or tuple targets, `break`, `continue` and `raise`, `from` clause inclu 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, and `del`. Not yet: `match`, `nonlocal` assignments, classes defined inside functions, -and a conditional expression or comprehension inside a `while` condition. Blocks inside a `try` +yet, `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` body carry exception edges to the handlers; `finally` is modelled on the normal path only. Methods of module-level classes are analysed like functions; other module-level code is skipped. An import inside a function is shown where it runs, as an `import` instruction naming the module, the bound name and the canonical symbol. A function using syntax outside this subset is reported by `--check` as diff --git a/src/coretrace_python/cfg/builder.py b/src/coretrace_python/cfg/builder.py index 0db932c..c144d7f 100644 --- a/src/coretrace_python/cfg/builder.py +++ b/src/coretrace_python/cfg/builder.py @@ -234,6 +234,8 @@ def statement( return self.conditional(node, block, continuation) if isinstance(node, nodes.While | nodes.For): return self.loop_statement(node, block, continuation) + if isinstance(node, nodes.Match): + return self.match_statement(node, block, continuation) if isinstance(node, nodes.With): return self.with_statement(node, block) if isinstance(node, nodes.Try): @@ -308,18 +310,63 @@ def loop_statement( ) -> _Open | None: header_id = self.new_id("loop") body_id = self.new_id("body") - exit_id = continuation if continuation is not None else self.new_id("exit") + after_id = continuation if continuation is not None else self.new_id("exit") + # An ``else`` clause runs when the loop is exhausted; ``break`` skips it. + exit_id = self.new_id("else") if node.orelse else after_id self.finish(block, Jump(header_id, node.span)) if isinstance(node, nodes.While): self.header(header_id, Branch(node.condition, body_id, exit_id, node.span)) else: self.header(header_id, ForEach(node.target, node.iterable, body_id, exit_id, node.span)) - self.loops.append((header_id, exit_id)) + self.loops.append((header_id, after_id)) try: self.sequence(node.body, _Open(body_id), header_id, node.span) finally: self.loops.pop() - return None if continuation is not None else _Open(exit_id) + if node.orelse: + self.sequence(node.orelse, _Open(exit_id), after_id, node.span) + return None if continuation is not None else _Open(after_id) + + def match_statement( + self, node: nodes.Match, block: _Open, continuation: BlockId | None + ) -> _Open | None: + """``match`` as an ``if`` chain over a hidden subject: literal, singleton, capture, + wildcard and or-patterns, with guards; other patterns are reported.""" + + subject = self.hidden("match", node.span) + statements: tuple[nodes.Statement, ...] = () + for case in reversed(node.cases): + condition, bindings = self.pattern(case.pattern, subject) + if case.guard is not None: + condition = nodes.BoolOp("and", (condition, case.guard), case.span) + statements = (*bindings, nodes.If(condition, case.body, statements, case.span)) + chain = (nodes.Assign(subject, node.subject, node.span), *statements) + return self.sequence(chain, block, continuation, node.span) + + def pattern(self, node: nodes.Pattern, subject: nodes.Name) -> tuple[nodes.Expression, tuple[nodes.Statement, ...]]: + """The condition a pattern tests on ``subject`` and the names it binds.""" + + if isinstance(node, nodes.ValuePattern): + return nodes.Compare("eq", subject, node.value, node.span), () + if isinstance(node, nodes.SingletonPattern): + return nodes.Compare("is", subject, nodes.Constant(node.value, node.span), node.span), () + if isinstance(node, nodes.WildcardPattern): + return nodes.Constant(True, node.span), () + if isinstance(node, nodes.CapturePattern): + binding = nodes.Assign(nodes.Name(node.name, node.span), subject, node.span) + if node.pattern is None: + return nodes.Constant(True, node.span), (binding,) + condition, inner = self.pattern(node.pattern, subject) + return condition, (*inner, binding) + if isinstance(node, nodes.OrPattern): + conditions: list[nodes.Expression] = [] + bindings: list[nodes.Statement] = [] + for alternative in node.alternatives: + condition, inner = self.pattern(alternative, subject) + conditions.append(condition) + bindings.extend(inner) + return nodes.BoolOp("or", tuple(conditions), node.span), tuple(bindings) + raise CFGError(f"{node.span.display()}: match pattern {node.kind} is not supported yet") def loop(self, keyword: str, span: SourceSpan) -> tuple[BlockId, BlockId]: if not self.loops: diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index ec8399e..c78f9b3 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/src/coretrace_python/frontend/ast_adapter.py @@ -270,12 +270,8 @@ def statement(self, node: ast.stmt) -> nodes.Statement: self.expression(node.test), self.block(node.body), self.block(node.orelse), span ) if isinstance(node, ast.While): - if node.orelse: - self.fail(node.orelse[0], "loop else clauses are not supported yet") - return nodes.While(self.expression(node.test), self.block(node.body), span) + return nodes.While(self.expression(node.test), self.block(node.body), span, self.block(node.orelse)) if isinstance(node, (ast.For, ast.AsyncFor)): - if node.orelse: - self.fail(node.orelse[0], "loop else clauses are not supported yet") body = self.block(node.body) if isinstance(node.target, ast.Name): target = nodes.Name(node.target.id, self.span(node.target)) @@ -290,11 +286,23 @@ def statement(self, node: ast.stmt) -> nodes.Statement: body, isinstance(node, ast.AsyncFor), span, + self.block(node.orelse), ) if isinstance(node, ast.Break): return nodes.Break(span) if isinstance(node, ast.Delete): return nodes.Delete(tuple(self.target(target) for target in node.targets), span) + if isinstance(node, ast.Match): + cases = tuple( + nodes.MatchCase( + self.pattern(case.pattern), + self.expression(case.guard) if case.guard is not None else None, + self.block(case.body), + self.span(case.pattern), + ) + for case in node.cases + ) + return nodes.Match(self.expression(node.subject), cases, span) if isinstance(node, ast.Continue): return nodes.Continue(span) if isinstance(node, ast.Raise): @@ -329,6 +337,21 @@ def statement(self, node: ast.stmt) -> nodes.Statement: return self.class_definition(node) self.fail(node) + def pattern(self, node: ast.pattern) -> nodes.Pattern: + span = self.span(node) + if isinstance(node, ast.MatchValue): + return nodes.ValuePattern(self.expression(node.value), span) + if isinstance(node, ast.MatchSingleton): + return nodes.SingletonPattern(node.value, span) + if isinstance(node, ast.MatchAs): + if node.name is None: + return nodes.WildcardPattern(span) + inner = self.pattern(node.pattern) if node.pattern is not None else None + return nodes.CapturePattern(node.name, inner, span) + if isinstance(node, ast.MatchOr): + return nodes.OrPattern(tuple(self.pattern(p) for p in node.patterns), span) + return nodes.UnsupportedPattern(type(node).__name__, span) + def block(self, statements: list[ast.stmt]) -> tuple[nodes.Statement, ...]: return tuple(found for statement in statements for found in self.statements(statement)) diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 531a250..250f5ec 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -266,6 +266,72 @@ class Delete: span: SourceSpan +@dataclass(frozen=True, slots=True) +class ValuePattern: + """``case ``: equality with a value.""" + + value: Expression + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class SingletonPattern: + """``case None`` / ``True`` / ``False``: identity with a singleton.""" + + value: object + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class WildcardPattern: + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class CapturePattern: + """``case name`` or ``case as name``: binds the subject.""" + + name: str + pattern: Pattern | None + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class OrPattern: + alternatives: tuple[Pattern, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class UnsupportedPattern: + """A sequence, mapping, class or star pattern; the CFG builder reports it.""" + + kind: str + span: SourceSpan + + +Pattern: TypeAlias = ( + ValuePattern | SingletonPattern | WildcardPattern | CapturePattern | OrPattern | UnsupportedPattern +) + + +@dataclass(frozen=True, slots=True) +class MatchCase: + pattern: Pattern + guard: Expression | None + body: tuple[Statement, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Match: + """``match subject:``; the CFG builder lays it out as an ``if`` chain.""" + + subject: Expression + cases: tuple[MatchCase, ...] + span: SourceSpan + + @dataclass(frozen=True, slots=True) class Assert: test: Expression @@ -286,6 +352,7 @@ class While: condition: Expression body: tuple[Statement, ...] span: SourceSpan + orelse: tuple[Statement, ...] = () @dataclass(frozen=True, slots=True) @@ -295,6 +362,7 @@ class For: body: tuple[Statement, ...] is_async: bool span: SourceSpan + orelse: tuple[Statement, ...] = () @dataclass(frozen=True, slots=True) @@ -431,6 +499,7 @@ class Class: | ExpressionStatement | Pass | Delete + | Match | Assert | If | While diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index b293687..cc72f2f 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -293,17 +293,22 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: self.expression(base, scope) scope.bind(node.name, BindingKind.CLASS, node.span) self.body(node.body, self.open(scope, ScopeKind.CLASS, node.name, node.span)) - elif isinstance(node, nodes.If): + elif isinstance(node, nodes.If | nodes.While): self.expression(node.condition, scope) self.body(node.body, scope) self.body(node.orelse, scope) - elif isinstance(node, nodes.While): - self.expression(node.condition, scope) - self.body(node.body, scope) elif isinstance(node, nodes.For): self.expression(node.iterable, scope) scope.bind(node.target.identifier, BindingKind.LOCAL, node.target.span) self.body(node.body, scope) + self.body(node.orelse, scope) + elif isinstance(node, nodes.Match): + self.expression(node.subject, scope) + for case in node.cases: + self.pattern(case.pattern, scope) + if case.guard is not None: + self.expression(case.guard, scope) + self.body(case.body, scope) elif isinstance(node, nodes.Raise): if node.exception is not None: self.expression(node.exception, scope) @@ -379,6 +384,17 @@ def expression(self, node: nodes.Expression, scope: _ScopeBuilder) -> None: else: raise TypeError(f"unknown expression: {node!r}") + def pattern(self, node: nodes.Pattern, scope: _ScopeBuilder) -> None: + if isinstance(node, nodes.ValuePattern): + self.expression(node.value, scope) + elif isinstance(node, nodes.CapturePattern): + scope.bind(node.name, BindingKind.LOCAL, node.span) + if node.pattern is not None: + self.pattern(node.pattern, scope) + elif isinstance(node, nodes.OrPattern): + for alternative in node.alternatives: + self.pattern(alternative, scope) + def target(self, node: nodes.Target, scope: _ScopeBuilder) -> None: if isinstance(node, nodes.Name): scope.bind(node.identifier, BindingKind.LOCAL, node.span) diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 790d273..494b381 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -61,7 +61,7 @@ ) from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeTable -from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolTable +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable from coretrace_python.source import SourceSpan from coretrace_python.taint.models import ( EntryPoint, @@ -375,12 +375,62 @@ def report( ) +Instances = Mapping[str, tuple[SymbolId, ...]] + + +def factory_instances( + module: nodes.Module, + scopes: ScopeTable, + symbols: SymbolTable, + summaries: SummaryTable, + project: SummaryIndex, +) -> Instances: + """Module-level names bound to the result of a project function whose summary + returns known symbols: ``app = create_app()`` is a ``flask.Flask`` like + ``app = Flask(__name__)``, so its decorators resolve.""" + + found: dict[str, tuple[SymbolId, ...]] = {} + module_scope = scopes.module_scope.id + for statement in module.body: + if not ( + isinstance(statement, nodes.Assign) + and isinstance(statement.target, nodes.Name) + and isinstance(statement.value, nodes.Call) + ): + continue + callee = statement.value.callee + externals: frozenset[SymbolId] = frozenset() + if isinstance(callee, nodes.Name) and callee.identifier in summaries.names: + externals = summaries.summary(callee.identifier).return_externals + else: + symbol = symbols.resolve_expression(module_scope, callee) + summary = project.summary(symbol) if symbol is not None else None + if summary is not None: + externals = summary.return_externals + if externals: + found[statement.target.identifier] = tuple(sorted(externals, key=str)) + return found + + +def _instance_symbols(expression: nodes.Expression, instances: Instances) -> tuple[SymbolId, ...]: + """The symbols an expression rooted at a factory instance may denote.""" + + if isinstance(expression, nodes.Name): + return instances.get(expression.identifier, ()) + if isinstance(expression, nodes.Attribute): + return tuple(s.attribute(expression.name) for s in _instance_symbols(expression.value, instances)) + if isinstance(expression, nodes.Call): + return _instance_symbols(expression.callee, instances) + return () + + def entry_point_of( function: nodes.Function, models: ModelTable, scopes: ScopeTable, symbols: SymbolTable, owner: nodes.Class | None = None, + instances: Instances | None = None, ) -> EntryPoint | None: """The entry-point model matching one of the function's decorators or, for a method, one of the bases of ``owner``, if any.""" @@ -397,8 +447,11 @@ def entry_point_of( enclosing_of = {} for expression in candidates: symbol = symbols.resolve_expression(enclosing_of.get(id(expression), enclosing), expression) - if symbol is not None: - entry = models.entry_point(symbol) + # ``app = create_app()`` resolves to the factory's symbol; the instance symbols + # say what the factory returns. + found = (*((symbol,) if symbol is not None else ()), *_instance_symbols(expression, instances or {})) + for candidate in found: + entry = models.entry_point(candidate) if entry is not None: return entry return None @@ -410,6 +463,7 @@ def parameter_sources( models: ModelTable, scopes: ScopeTable, symbols: SymbolTable, + instances: Instances | None = None, ) -> Mapping[int, Source]: """The attacker-controlled parameters of ``function``, by index: every parameter of an entry point (``self`` excepted for a method) and every parameter annotated with a @@ -420,7 +474,7 @@ def parameter_sources( None, ) sources: dict[int, Source] = {} - entry = entry_point_of(function, models, scopes, symbols, owner) + entry = entry_point_of(function, models, scopes, symbols, owner, instances) if entry is not None: first = 1 if owner is not None else 0 for index in range(first, len(function.parameters)): @@ -490,7 +544,18 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> TaintFacts: graph, ctx.get(SummaryAnalysis), parameter_sources( - function, ctx.module, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis) + function, + ctx.module, + models, + ctx.get(ScopeAnalysis), + ctx.get(SymbolAnalysis), + factory_instances( + ctx.module, + ctx.get(ScopeAnalysis), + ctx.get(SymbolAnalysis), + ctx.get(SummaryAnalysis), + ctx.get(ProjectSummaries), + ), ), ctx.get(ProjectSummaries), ctx.get(HeapAnalysis, function),