From 11866f68a08ce6a580702b075c2f190445f60826 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 06:30:19 +0200 Subject: [PATCH 1/2] test: define item symbols, session authorization, credential names and del --- tests/test_real_code_precision.py | 183 ++++++++++++++++++++++++++++++ 1 file changed, 183 insertions(+) create mode 100644 tests/test_real_code_precision.py diff --git a/tests/test_real_code_precision.py b/tests/test_real_code_precision.py new file mode 100644 index 0000000..51db053 --- /dev/null +++ b/tests/test_real_code_precision.py @@ -0,0 +1,183 @@ +"""Acceptance tests for precision gaps found by hand on the ``tests-project/`` repositories. + +Reading the two true path traversals of ``uploadbox-cloud-storage-solution`` showed four +things around them: a third traversal missed because ``request.files['file'].save(...)`` +resolves no symbol, a ``'logged_in' in session`` check the refutation engine did not +count as authorization, a secret assigned through ``app.config['SECRET_KEY']`` the secret +detector did not name, and a ``del`` statement that blocked a whole file elsewhere. + +Expected to remain red until item symbols, membership authorization, subscript and +attribute credential names and ``del`` exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Confidence, Finding +from coretrace_python.frontend import build_hir +from coretrace_python.interprocedural import CallGraphAnalysis +from coretrace_python.ir.lowering import lower_module +from coretrace_python.ir.printer import format_module +from coretrace_python.plugins.secrets import literals +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager + +try: + from coretrace_python.hir.nodes import Delete + from coretrace_python.ir.model import DelAttr, DelItem +except ImportError as error: # pragma: no cover - red until del lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_precision() -> None: + if MISSING is not None: + pytest.fail(f"real-code precision pass is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("app.py", text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, int]]: + return sorted((f.rule_id, f.span.start_line) for f in findings) + + +FLASK = "import os\nfrom flask import Flask, request, session, send_file\n\napp = Flask(__name__)\n\n" + + +# --------------------------------------------------------------------------- item symbols + + +def test_items_of_symbol_containers_carry_the_container_symbol() -> None: + module = build_hir( + SourceManager().add_source( + "m.py", "from flask import request\n\ndef f():\n stored = request.files['file']\n stored.save('x')\n" + ) + ) + graph = engine.build_manager(module).get(CallGraphAnalysis) + (site,) = graph.sites("f") + + from coretrace_python.interprocedural import ExternalSymbol + + assert isinstance(site.target, ExternalSymbol) + assert site.target.symbol == SymbolId("python.flask.request.files.save") + + +def test_uploaded_files_saved_under_their_own_name_are_path_traversals() -> None: + findings = check( + FLASK + "@app.route('/upload', methods=['POST'])\n" + "def upload():\n" + " f = request.files['file']\n" + " f.save(os.path.join('/srv/files', session['email'], f.filename))\n" + " return 'ok'\n" + ) + assert rules(findings) == [("path-traversal", 9)] + + +def test_secure_filename_sanitises_the_upload_name() -> None: + findings = check( + FLASK + "from werkzeug.utils import secure_filename\n\n" + "@app.route('/upload', methods=['POST'])\n" + "def upload():\n" + " f = request.files['file']\n" + " f.save(os.path.join('/srv/files', secure_filename(f.filename)))\n" + " return 'ok'\n" + ) + assert findings == () + + +# --------------------------------------------------------------------------- session authorization + + +def test_membership_in_the_session_is_an_authorization_guard() -> None: + findings = check( + FLASK + "@app.route('/delete', methods=['POST'])\n" + "def delete():\n" + " name = request.get_json()['filename']\n" + " if 'logged_in' in session:\n" + " os.remove(os.path.join('/srv/files', name))\n" + " return 'ok'\n" + ) + (finding,) = findings + assert finding.rule_id == "path-traversal" + assert finding.confidence is Confidence.MEDIUM + assert finding.metadata["verdict"] == "hotspot" + assert finding.metadata["evidence"] == "behind authorization (session) at line 9" + + +def test_early_return_on_missing_session_counts_too() -> None: + findings = check( + FLASK + "@app.route('/delete', methods=['POST'])\n" + "def delete():\n" + " if 'logged_in' not in session:\n" + " return 'no'\n" + " os.remove(request.get_json()['filename'])\n" + ) + (finding,) = findings + assert finding.metadata["verdict"] == "hotspot" + + +def test_unrelated_membership_checks_are_ordinary_guards() -> None: + findings = check( + FLASK + "@app.route('/delete', methods=['POST'])\n" + "def delete():\n" + " name = request.get_json()['filename']\n" + " if 'x' in name:\n" + " os.remove(name)\n" + ) + (finding,) = findings + assert finding.metadata["verdict"] == "hotspot" + assert "authorization" not in finding.metadata["evidence"] + + +# --------------------------------------------------------------------------- credential names + + +def test_credentials_bound_through_subscripts_and_attributes_are_named() -> None: + module = build_hir( + SourceManager().add_source( + "settings.py", + "app.config['SECRET_KEY'] = 'find@me1290'\n" + "class Client:\n def __init__(self):\n self.password = 'hunter2'\n" + "options[0] = 'positional'\n", + ) + ) + + found = [(value, name) for value, name, _, _ in literals(module)] + + assert found == [("find@me1290", "SECRET_KEY"), ("hunter2", "password"), ("positional", None)] + findings = check("app = object()\napp.config['SECRET_KEY'] = 'find@me1290'\n") + assert [(f.rule_id, f.metadata["name"]) for f in findings] == [("hardcoded-credential", "SECRET_KEY")] + + +# --------------------------------------------------------------------------- del + + +def test_del_statements_lower_to_delete_effects() -> None: + module = build_hir( + SourceManager().add_source("d.py", "def f(d, obj, x):\n del d['k'], obj.attr\n del x\n return d\n") + ) + statement = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(statement, Delete) and len(statement.targets) == 2 + + function = lower_module(module).functions[0] + effects = [i for b in function.blocks for i in b.instructions if isinstance(i, DelItem | DelAttr)] + assert [type(i).__name__ for i in effects] == ["DelItem", "DelAttr"] + text = format_module(lower_module(module)) + assert "del_item" in text and 'del_attr %1, "attr"' in text + + +def test_del_does_not_block_a_file() -> None: + findings = check("import os\n\ndef f(cache):\n del cache['x']\n os.system(input())\n") + assert rules(findings) == [("command-injection", 5)] From e1658d186f9997ab05f43a0fd293260ebbb79828 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 06:30:19 +0200 Subject: [PATCH 2/2] fix: resolve item symbols, model uploads and session checks, name subscript credentials, lower del --- README.md | 7 +++--- plugins/models/flask/flask_models.py | 3 +++ src/coretrace_python/findings/refutation.py | 6 +++++ src/coretrace_python/frontend/ast_adapter.py | 2 ++ src/coretrace_python/hir/nodes.py | 7 ++++++ .../interprocedural/callgraph.py | 22 +++++++++++++++---- src/coretrace_python/ir/lowering.py | 11 ++++++++++ src/coretrace_python/ir/model.py | 20 ++++++++++++++++- src/coretrace_python/ir/printer.py | 6 +++++ src/coretrace_python/plugins/secrets.py | 15 +++++++++++-- src/coretrace_python/semantic/scopes.py | 3 +++ 11 files changed, 92 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 86159c8..05c5186 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,8 @@ component per requirement with its package URL, and the advisories affecting the vulnerabilities. The shipped `secrets/` plugin scans every string literal of the Python sources for provider-specific secret formats (AWS, GitHub, Slack, Stripe, Google, private keys, JWTs, SendGrid, Twilio), for credential-like names bound to a real value -(`password`, `token`, `api_key` and the like, placeholders excluded) and for opaque +(`password`, `token`, `api_key` and the like, bound by name, attribute or constant key +such as `app.config['SECRET_KEY']`, placeholders excluded) and for opaque high-entropy tokens; one finding per literal, at high, medium and low confidence respectively, with a redacted preview and never the secret itself. Other plugins add providers by subclassing `SecretDetector` with their own patterns. Configuration files @@ -144,8 +145,8 @@ 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. Not yet: `match`, `del`, `nonlocal` assignments, classes defined inside functions, and -a conditional expression or comprehension inside a `while` condition. Blocks inside a `try` +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` 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/plugins/models/flask/flask_models.py b/plugins/models/flask/flask_models.py index bd7dd43..761456e 100644 --- a/plugins/models/flask/flask_models.py +++ b/plugins/models/flask/flask_models.py @@ -37,9 +37,12 @@ class FlaskModels(ModelPlugin): Sink(_sym("flask.Response"), TaintKind.HTML), Sink(_sym("flask.Markup"), TaintKind.HTML), Sink(_sym("flask.send_file"), TaintKind.PATH), + Sink(_sym("flask.request.files.save"), TaintKind.PATH), + Sanitizer(_sym("werkzeug.utils.secure_filename"), TaintKind.PATH), Sanitizer(_sym("flask.escape"), TaintKind.HTML), Sanitizer(_sym("markupsafe.escape"), TaintKind.HTML), AuthorizationGuard(_sym("flask_login.login_required"), "login"), AuthorizationGuard(_sym("flask_login.fresh_login_required"), "login"), AuthorizationGuard(_sym("flask_login.current_user.is_authenticated"), "login"), + AuthorizationGuard(_sym("flask.session"), "session"), ) diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index 3efff39..8acec45 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -289,6 +289,12 @@ def authorized_by(self, condition: Value, truth: bool) -> AuthorizationGuard | N if found is not None: return found return None + if isinstance(definition, Compare) and definition.operator in ("in", "not_in"): + # ``'logged_in' in session``: membership in an authorization store. + if truth != (definition.operator == "in"): + return None + container = self.symbols.get(definition.right) + return self.models.authorization(container) if container is not None else None if not truth: return None symbol = self.symbols.get(condition) diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 4d97baa..ec8399e 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/src/coretrace_python/frontend/ast_adapter.py @@ -293,6 +293,8 @@ def statement(self, node: ast.stmt) -> nodes.Statement: ) 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.Continue): return nodes.Continue(span) if isinstance(node, ast.Raise): diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 3171748..531a250 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -260,6 +260,12 @@ class Pass: span: SourceSpan +@dataclass(frozen=True, slots=True) +class Delete: + targets: tuple[Target, ...] + span: SourceSpan + + @dataclass(frozen=True, slots=True) class Assert: test: Expression @@ -424,6 +430,7 @@ class Class: | Return | ExpressionStatement | Pass + | Delete | Assert | If | While diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 53429fb..c26ccab 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -17,7 +17,16 @@ from coretrace_python.cfg import CFGError from coretrace_python.hir import nodes from coretrace_python.ir.lowering import LoweringError, analyzable_functions, qualified_name -from coretrace_python.ir.model import Call, FunctionIR, GetAttr, Global, Symbol, Value, WithEnter +from coretrace_python.ir.model import ( + Call, + FunctionIR, + GetAttr, + GetItem, + Global, + Symbol, + Value, + WithEnter, +) from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.semantic.scopes import BindingKind, ScopeAnalysis, ScopeTable from coretrace_python.semantic.symbols import SymbolId @@ -100,9 +109,10 @@ def callers(self, name: str) -> frozenset[str]: def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]: - """Symbols of values: ``Symbol`` results, attributes of symbol values, results of - calling a symbol (``sqlite3.connect(p)`` denotes ``python.sqlite3.connect``) and the - values a ``with`` on such a result binds. Known functions and parameters derive nothing.""" + """Symbols of values: ``Symbol`` results, attributes and items of symbol values, + results of calling a symbol (``sqlite3.connect(p)`` denotes ``python.sqlite3.connect``) + and the values a ``with`` on such a result binds. Known functions and parameters + derive nothing.""" symbols: dict[Value, SymbolId] = {} changed = True @@ -117,6 +127,10 @@ def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]: symbol = instruction.symbol_id elif isinstance(instruction, GetAttr) and instruction.object in symbols: symbol = symbols[instruction.object].attribute(instruction.attribute) + elif isinstance(instruction, GetItem) and instruction.object in symbols: + # An item of a symbol-denoted container (``request.files['f']``) + # carries the container's symbol, so its methods resolve. + symbol = symbols[instruction.object] elif isinstance(instruction, Call | WithEnter): origin = ( instruction.callee if isinstance(instruction, Call) else instruction.context diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index fe51b5e..30fe373 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -33,6 +33,8 @@ Catch, Compare, Constant, + DelAttr, + DelItem, EffectInstruction, ForNext, FunctionIR, @@ -237,6 +239,15 @@ def statement(self, node: nodes.Statement) -> None: if isinstance(node, nodes.Assign): self.store(node.target, self.expression(node.value)) return + if isinstance(node, nodes.Delete): + for target in node.targets: + if isinstance(target, nodes.Subscript): + obj, key = self.expression(target.value), self.expression(target.key) + self.emit_effect(DelItem(None, target.span, obj, key)) + elif isinstance(target, nodes.Attribute): + self.emit_effect(DelAttr(None, target.span, self.expression(target.value), target.name)) + # ``del name`` unbinds a local; the SSA form has no slot to clear. + return if isinstance(node, nodes.Function): # A nested definition is a value bound to its name; its body is its own # scope and is not lowered here. diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index de8c44a..a1ccc88 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -311,6 +311,22 @@ class Assert(Operands): message: Value | None +@dataclass(frozen=True) +class DelItem(Operands): + result: None + location: SourceSpan + object: Value + key: Value + + +@dataclass(frozen=True) +class DelAttr(Operands): + result: None + location: SourceSpan + object: Value + attribute: str + + @dataclass(frozen=True) class SetGlobal(Operands): """Assignment to a name declared ``global``.""" @@ -360,7 +376,9 @@ class Import(Operands): | Await | Yield ) -EffectInstruction: TypeAlias = StoreLocal | SetAttr | SetItem | WithExit | Assert | Import | SetGlobal +EffectInstruction: TypeAlias = ( + StoreLocal | SetAttr | SetItem | WithExit | Assert | Import | SetGlobal | DelItem | DelAttr +) Instruction: TypeAlias = ValueInstruction | EffectInstruction diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index 6aa7b25..94d8124 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -16,6 +16,8 @@ Catch, Compare, Constant, + DelAttr, + DelItem, ForNext, GetAttr, GetItem, @@ -93,6 +95,10 @@ def _instruction(instruction: Instruction) -> str: return f"{_value(instruction.result)} = build_set {', '.join(parts)}".rstrip() if isinstance(instruction, MakeFunction): return f'{_value(instruction.result)} = make_function "{instruction.name}"' + if isinstance(instruction, DelItem): + return f"del_item {_value(instruction.object)}, {_value(instruction.key)}" + if isinstance(instruction, DelAttr): + return f'del_attr {_value(instruction.object)}, "{instruction.attribute}"' if isinstance(instruction, SetGlobal): return f'set_global "{instruction.name}", {_value(instruction.value)}' if isinstance(instruction, BuildString): diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 2d73cef..675a554 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -66,8 +66,7 @@ def _walk(node: Node, name: str | None, function: str | None) -> Iterator[Litera yield from _walk(child, None, None) return if isinstance(node, nodes.Assign): - target = node.target.identifier if isinstance(node.target, nodes.Name) else None - yield from _walk(node.value, target, function) + yield from _walk(node.value, _bound_name(node.target), function) return if isinstance(node, nodes.Keyword): yield from _walk(node.value, node.name, function) @@ -87,6 +86,18 @@ def _walk(node: Node, name: str | None, function: str | None) -> Iterator[Litera yield from _walk(child, None, function) +def _bound_name(target: nodes.Target) -> str | None: + """The name an assignment binds: ``name``, ``obj.attr`` or ``mapping['key']``.""" + + if isinstance(target, nodes.Name): + return target.identifier + if isinstance(target, nodes.Attribute): + return target.name + if isinstance(target, nodes.Subscript) and isinstance(target.key, nodes.Constant): + return target.key.value if isinstance(target.key.value, str) else None + return None + + @dataclass(frozen=True) class SecretPattern: """A provider-specific secret format.""" diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index 8afda18..b293687 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -309,6 +309,9 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: self.expression(node.exception, scope) if node.cause is not None: self.expression(node.cause, scope) + elif isinstance(node, nodes.Delete): + for target in node.targets: + self.target(target, scope) elif not isinstance(node, nodes.Pass | nodes.Break | nodes.Continue): raise TypeError(f"unknown statement: {node!r}")