From b71384c689e046f6d9e4470fb15df1b99fd8626e Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 19:25:00 -0700 Subject: [PATCH 1/5] test(analyzer): define shell truthiness core contract Signed-off-by: Christopher Kevin --- .../nodes/analyzers/test_shared_python_ast.py | 6 +- .../analyzers/test_tool_misuse_python_ast.py | 251 ++++++++++++++++++ tests/nodes/test_security_end_to_end.py | 75 ++++++ 3 files changed, 331 insertions(+), 1 deletion(-) create mode 100644 tests/nodes/analyzers/test_tool_misuse_python_ast.py diff --git a/tests/nodes/analyzers/test_shared_python_ast.py b/tests/nodes/analyzers/test_shared_python_ast.py index fa864ed60..1b152dfbe 100644 --- a/tests/nodes/analyzers/test_shared_python_ast.py +++ b/tests/nodes/analyzers/test_shared_python_ast.py @@ -16,6 +16,7 @@ behavioral_taint_tracking, static_patterns_data_exfiltration, static_patterns_output_handling, + static_patterns_tool_misuse, ) from skillspector.nodes.build_context import build_context from skillspector.nodes.deduplicate import deduplicate @@ -62,7 +63,8 @@ def test_preparsed_python_is_reused_by_all_ast_analyzers(tmp_path, monkeypatch) "import subprocess\n" "payload = input()\n" "environment = os.environ.copy()\n" - "subprocess.run(output)\n" + "enabled = True\n" + "subprocess.run(output, shell=enabled)\n" "exec(payload)\n", encoding="utf-8", ) @@ -90,11 +92,13 @@ def count_parse(*args, **kwargs): data_findings = static_patterns_data_exfiltration.node(state)["findings"] output_findings = static_patterns_output_handling.node(state)["findings"] + tool_misuse_findings = static_patterns_tool_misuse.node(state)["findings"] ast_findings = behavioral_ast.node(state)["findings"] taint_findings = behavioral_taint_tracking.node(state)["findings"] assert any(finding.rule_id == "E2" for finding in data_findings) assert any(finding.rule_id == "OH1" for finding in output_findings) + assert any(finding.rule_id == "TM1" for finding in tool_misuse_findings) assert any(finding.rule_id == "AST1" for finding in ast_findings) assert any(finding.rule_id == "TT5" for finding in taint_findings) assert parse_calls == 1 diff --git a/tests/nodes/analyzers/test_tool_misuse_python_ast.py b/tests/nodes/analyzers/test_tool_misuse_python_ast.py new file mode 100644 index 000000000..64edd9841 --- /dev/null +++ b/tests/nodes/analyzers/test_tool_misuse_python_ast.py @@ -0,0 +1,251 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Focused coverage for issue #475's ordinary-Python binding form.""" + +from __future__ import annotations + +import pytest + +from skillspector.nodes.analyzers import static_patterns_tool_misuse as tm_module + + +def _run(content: str, path: str = "run.py") -> dict: + return tm_module.node({"components": [path], "file_cache": {path: content}}) + + +def _tm1(content: str, path: str = "run.py") -> list: + return [finding for finding in _run(content, path)["findings"] if finding.rule_id == "TM1"] + + +@pytest.mark.parametrize( + "value", + [ + pytest.param("True", id="issue-body-boolean"), + pytest.param("'True'", id="reporter-attachment-string"), + ], +) +def test_issue_475_multiline_binding_matches_direct_tm1(value: str) -> None: + findings = _tm1( + "import subprocess\n" + "command = f'python a.py'\n" + f"enabled = {value}\n" + "result = subprocess.run(\n" + " command,\n" + " shell=enabled,\n" + " capture_output=True,\n" + " text=True,\n" + ")\n" + ) + + assert len(findings) == 1 + assert findings[0].start_line == 4 + assert findings[0].severity == "HIGH" + assert findings[0].confidence == pytest.approx(0.9) + assert "shell=enabled" in findings[0].matched_text + assert not findings[0].evidence + + +@pytest.mark.parametrize( + "value", + [ + pytest.param("1", id="integer"), + pytest.param("-1", id="negative-integer"), + pytest.param("(0,)", id="nonempty-tuple"), + pytest.param("not False", id="negation"), + ], +) +def test_simple_immutable_truthy_values_are_tracked(value: str) -> None: + assert ( + len(_tm1(f"import subprocess\nenabled = {value}\nsubprocess.run(cmd, shell=enabled)\n")) + == 1 + ) + + +@pytest.mark.parametrize( + "value", + [ + pytest.param("False", id="boolean"), + pytest.param("0", id="integer"), + pytest.param("''", id="string"), + pytest.param("()", id="tuple"), + pytest.param("None", id="none"), + ], +) +def test_definitely_false_values_are_not_tracked(value: str) -> None: + assert not _tm1(f"import subprocess\nenabled = {value}\nsubprocess.run(cmd, shell=enabled)\n") + + +def test_simple_alias_chain_and_bare_popen_are_tracked() -> None: + findings = _tm1( + "first = 'enabled'\nsecond = first\nthird = second\nPopen(command, shell=third)\n" + ) + + assert len(findings) == 1 + assert findings[0].start_line == 4 + + +@pytest.mark.parametrize( + "rebind", + [ + pytest.param("enabled = False", id="false-assignment"), + pytest.param("enabled = dynamic", id="unknown-assignment"), + pytest.param("import pathlib as enabled", id="import"), + pytest.param("from settings import enabled", id="from-import"), + ], +) +def test_rebinding_invalidates_truthy_fact(rebind: str) -> None: + assert not _tm1( + f"import subprocess\nenabled = True\n{rebind}\nsubprocess.run(command, shell=enabled)\n" + ) + + +def test_import_side_effect_boundary_clears_truth_facts() -> None: + assert not _tm1( + "import subprocess\nenabled = True\nimport attacker\n" + "subprocess.run(command, shell=enabled)\n" + ) + + +@pytest.mark.parametrize( + "shadow", + [ + pytest.param("subprocess = Proxy()", id="assignment"), + pytest.param("import other as subprocess", id="import-alias"), + pytest.param("for subprocess in values:\n pass", id="compound-binder"), + pytest.param("subprocess.run = Proxy()", id="attribute-mutation"), + pytest.param("subprocess, other = pair", id="unpacking"), + ], +) +def test_explicit_subprocess_shadow_rejects_bound_call(shadow: str) -> None: + assert not _tm1(f"{shadow}\nenabled = True\nsubprocess.run(cmd, shell=enabled)\n") + + +def test_explicit_import_reestablishes_direct_receivers() -> None: + assert ( + len( + _tm1( + "subprocess = Proxy()\n" + "import subprocess\n" + "Popen = Proxy()\n" + "from subprocess import Popen\n" + "enabled = True\n" + "subprocess.run(command, shell=enabled)\n" + "Popen(command, shell=enabled)\n" + ) + ) + == 2 + ) + + +def test_relative_import_does_not_establish_bare_popen() -> None: + assert not _tm1( + "Popen = proxy\nfrom .subprocess import Popen\nenabled = True\n" + "Popen(command, shell=enabled)\n" + ) + + +def test_function_local_binding_and_outer_fact_are_independent() -> None: + findings = _tm1( + "outer = True\n" + "def execute(command):\n" + " enabled = 'True'\n" + " subprocess.run(command, shell=enabled)\n" + "subprocess.run(command, shell=outer)\n" + ) + + assert [finding.start_line for finding in findings] == [4, 5] + + +def test_function_compile_time_receiver_shadow_rejects_earlier_lookup() -> None: + assert not _tm1( + "def execute(command):\n" + " enabled = True\n" + " subprocess.run(command, shell=enabled)\n" + " subprocess = Proxy()\n" + ) + + +def test_later_global_receiver_mutation_suppresses_function_body() -> None: + assert not _tm1( + "def execute(command):\n" + " enabled = True\n" + " subprocess.run(command, shell=enabled)\n" + "subprocess = Proxy()\n" + ) + + +def test_passive_function_definition_preserves_outer_fact() -> None: + assert ( + len( + _tm1( + "enabled = True\n" + "def helper(value=1):\n" + " pass\n" + "subprocess.run(command, shell=enabled)\n" + ) + ) + == 1 + ) + + +@pytest.mark.parametrize( + "compound", + [ + pytest.param("if condition:\n pass", id="if"), + pytest.param("for item in values:\n pass", id="for"), + pytest.param("with provider():\n pass", id="with"), + pytest.param("try:\n pass\nexcept Exception:\n pass", id="try"), + pytest.param("class Local:\n pass", id="class"), + ], +) +def test_compound_statement_conservatively_clears_truth_facts(compound: str) -> None: + assert not _tm1(f"enabled = True\n{compound}\nsubprocess.run(command, shell=enabled)\n") + + +def test_calls_inside_compound_statements_are_out_of_scope() -> None: + assert not _tm1( + "if condition:\n enabled = True\n subprocess.run(command, shell=enabled)\n" + ) + + +@pytest.mark.parametrize( + "argument", + [ + pytest.param("disable()", id="call"), + pytest.param("mutator.command", id="attribute"), + pytest.param("holder[0]", id="subscript"), + pytest.param("left + right", id="operator"), + pytest.param("f'{value}'", id="formatted-string"), + pytest.param("[item for item in items]", id="comprehension"), + pytest.param("*commands", id="starred-expansion"), + ], +) +def test_side_effect_capable_call_arguments_are_rejected(argument: str) -> None: + assert not _tm1(f"enabled = True\nsubprocess.run({argument}, shell=enabled)\n") + + +def test_unsupported_assignment_clears_existing_facts() -> None: + assert not _tm1("enabled = True\nresult = factory()\nsubprocess.run(cmd, shell=enabled)\n") + + +def test_annotated_assignment_is_outside_side_effect_free_contract() -> None: + assert not _tm1("enabled: bool = True\nsubprocess.run(command, shell=enabled)\n") + + +def test_assignment_rhs_direct_call_is_inspected_before_invalidation() -> None: + findings = _tm1("enabled = True\nresult = subprocess.run(command, shell=enabled)\n") + + assert len(findings) == 1 + assert findings[0].start_line == 2 + + +def test_true_prefixed_identifier_has_one_lexical_owner() -> None: + findings = _tm1("true_value = True\nsubprocess.run(command, shell=true_value)\n") + + assert len(findings) == 1 + + +@pytest.mark.parametrize("path", ["run.pyw", "run", "run.sh"]) +def test_non_py_surfaces_do_not_enable_ast_companion(path: str) -> None: + assert not _tm1("enabled = True\nsubprocess.run(command, shell=enabled)\n", path) diff --git a/tests/nodes/test_security_end_to_end.py b/tests/nodes/test_security_end_to_end.py index 76d47965d..d6bbdd81f 100644 --- a/tests/nodes/test_security_end_to_end.py +++ b/tests/nodes/test_security_end_to_end.py @@ -297,6 +297,81 @@ async def _assert_incomplete_across_public_surfaces( assert sc9["evidence"]["excluded_inspection_incomplete"] is True +@pytest.mark.parametrize( + "bound_value", + [ + pytest.param("True", id="boolean"), + pytest.param("'True'", id="reporter-truthy-string"), + ], +) +def test_tm1_bound_true_matches_literal_in_graph( + tmp_path: Path, + bound_value: str, +) -> None: + direct = tmp_path / "direct-shell" + bound = tmp_path / "bound-shell" + _write_bundle( + direct, + { + "SKILL.md": "# Shell helper", + "run.py": "import subprocess\nsubprocess.run(command, shell=True)\n", + }, + ) + _write_bundle( + bound, + { + "SKILL.md": "# Shell helper", + "run.py": ( + "import subprocess\n" + f"use_shell = {bound_value}\n" + "subprocess.run(command, shell=use_shell)\n" + ), + }, + ) + + direct_result = _scan(direct) + bound_result = _scan(bound) + direct_tm1 = _assert_rule(direct_result, "TM1", "run.py") + bound_tm1 = _assert_rule(bound_result, "TM1", "run.py") + + assert len(direct_tm1) == len(bound_tm1) == 1 + assert (bound_tm1[0].severity, bound_tm1[0].confidence) == ( + direct_tm1[0].severity, + direct_tm1[0].confidence, + ) + assert ( + bound_result["risk_score"], + bound_result["risk_severity"], + bound_result["risk_recommendation"], + ) == ( + direct_result["risk_score"], + direct_result["risk_severity"], + direct_result["risk_recommendation"], + ) + + +@pytest.mark.asyncio +async def test_tm1_bound_true_across_public_surfaces(tmp_path: Path) -> None: + bound = tmp_path / "bound-shell-public" + _write_bundle( + bound, + { + "SKILL.md": "# Shell helper", + "run.py": ( + "import subprocess\nuse_shell = True\nsubprocess.run(command, shell=use_shell)\n" + ), + }, + ) + + result = _scan(bound) + _assert_rule(result, "TM1", "run.py") + await _assert_rules_across_public_surfaces( + bound, + expected_locations={"TM1": {"run.py"}}, + python_result=result, + ) + + @pytest.mark.parametrize( ("finding", "normal_files", "bypass_files", "rule_id", "normal_path", "bypass_path"), [ From 96685159ff9cf04a71354c38771c9aeb33dcc3b4 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 19:32:47 -0700 Subject: [PATCH 2/5] fix(analyzer): detect bound shell truthiness Signed-off-by: Christopher Kevin --- .../analyzers/static_patterns_tool_misuse.py | 71 +- .../static_python_shell_truthiness.py | 666 ++++++++++++++++++ 2 files changed, 722 insertions(+), 15 deletions(-) create mode 100644 src/skillspector/nodes/analyzers/static_python_shell_truthiness.py diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index 6d8876a22..1d6a8da65 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -2550,6 +2550,26 @@ def _line_containing(content: str, start: int, end: int) -> str: return content[line_start:line_end] +def _classify_tm1( + context: str, + matched_text: str, + matched_line: str, + confidence: float, + file_type: str, +) -> tuple[Severity, float]: + """Apply the existing TM1 contextual classification to one candidate.""" + if ( + _is_safe_container_command(context) + or _is_safe_dockerfile_idiom(context, matched_text) + or _is_safe_cache_cleanup(matched_line) + ): + return Severity.LOW, min(confidence, 0.15) + adjusted = ( + min(1.0, confidence + 0.1) if file_type in ("python", "shell", "javascript") else confidence + ) + return Severity.HIGH, adjusted + + def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: """Analyze content for tool misuse patterns (TM1–TM3).""" findings: list[AnalyzerFinding] = [] @@ -2569,20 +2589,13 @@ def ctx(start: int) -> str: matched = matched_text[:200] matched_line = _line_containing(content, match_start, match_end) - if ( - _is_safe_container_command(context_text) - or _is_safe_dockerfile_idiom(context_text, matched) - or _is_safe_cache_cleanup(matched_line) - ): - adj = min(confidence, 0.15) - sev = Severity.LOW - else: - adj = ( - min(1.0, confidence + 0.1) - if file_type in ("python", "shell", "javascript") - else confidence - ) - sev = Severity.HIGH + sev, adj = _classify_tm1( + context_text, + matched, + matched_line, + confidence, + file_type, + ) candidate_key = (line_num, " ".join(matched.strip().split())) existing = tm1_findings_by_key.get(candidate_key) if existing is not None: @@ -2667,6 +2680,34 @@ def ctx(start: int) -> str: def node(state: SkillspectorState) -> AnalyzerNodeResponse: """Run tool_misuse patterns and return findings.""" - response = static_runner.run_static_patterns_with_ledger(state, [sys.modules[__name__]]) + from . import static_python_shell_truthiness + + response = static_runner.run_static_patterns_with_ledger( + state, + [sys.modules[__name__], static_python_shell_truthiness], + ) + file_cache = state.get("file_cache", {}) + for finding in response["findings"]: + if ( + finding.evidence.pop( + static_python_shell_truthiness.BOUND_SHELL_EVIDENCE, + None, + ) + is not True + ): + continue + content = file_cache.get(finding.file, "") + content_lines = content.splitlines() + line_index = max(0, finding.start_line - 1) + matched = (finding.matched_text or "")[:200] + matched_line = content_lines[line_index] if line_index < len(content_lines) else matched + severity, finding.confidence = _classify_tm1( + finding.context or "", + matched, + matched_line, + finding.confidence, + "python", + ) + finding.severity = severity.value logger.info("%s: %d findings", ANALYZER_ID, len(response["findings"])) return response diff --git a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py new file mode 100644 index 000000000..741a9d544 --- /dev/null +++ b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py @@ -0,0 +1,666 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Find direct subprocess calls using a definitely truthy local name. + +This companion recognizes the straight-line ordinary-Python form reported in +issue #475. Call arguments must be passive, and unsupported expressions or +compound statements discard facts rather than guessing about Python execution. +""" + +from __future__ import annotations + +import ast + +from skillspector.models import AnalyzerFinding, Location, Severity +from skillspector.python_ast import ParsedPythonFile, parse_python_source + +from .common import get_context_from_lines, get_source_segment +from .pattern_defaults import PatternCategory + +ANALYZER_ID = "static_patterns_tool_misuse" +USES_PYTHON_AST = True +BOUND_SHELL_EVIDENCE = "_tm1_bound_shell_value" +_DIRECT_CALL_NAMES = frozenset({"subprocess", "Popen"}) + + +def _truth_value( + expression: ast.expr | None, + facts: dict[str, bool], +) -> bool | None: + """Return truth for a small, immutable, side-effect-free expression subset.""" + if expression is None: + return None + + resolved: dict[ast.expr, bool] = {} + pending: list[tuple[ast.expr, bool]] = [(expression, False)] + while pending: + current, expanded = pending.pop() + if isinstance(current, ast.Constant): + resolved[current] = bool(current.value) + elif isinstance(current, ast.Name): + if current.id not in facts: + return None + resolved[current] = facts[current.id] + elif isinstance(current, ast.Tuple): + if not current.elts: + resolved[current] = False + elif any(isinstance(item, ast.Starred) for item in current.elts): + return None + elif all(_is_passive_argument(item) for item in current.elts): + resolved[current] = True + else: + return None + elif isinstance(current, ast.UnaryOp): + if not isinstance(current.op, ast.Not) and not ( + isinstance(current.op, (ast.UAdd, ast.USub)) + and isinstance(current.operand, ast.Constant) + and type(current.operand.value) in (bool, int, float, complex) + ): + return None + if expanded: + operand = resolved[current.operand] + resolved[current] = not operand if isinstance(current.op, ast.Not) else operand + else: + pending.append((current, True)) + pending.append((current.operand, False)) + else: + return None + return resolved[expression] + + +def _update_trusted_names_from_import( + statement: ast.Import | ast.ImportFrom, + trusted_names: set[str], +) -> None: + """Update only direct receiver names that the import actually binds.""" + if isinstance(statement, ast.Import): + for imported in statement.names: + bound = imported.asname or imported.name.partition(".")[0] + if imported.name == "subprocess" and bound == "subprocess": + trusted_names.add(bound) + elif bound in trusted_names: + trusted_names.discard(bound) + return + + if any(imported.name == "*" for imported in statement.names): + trusted_names.clear() + return + for imported in statement.names: + bound = imported.asname or imported.name + if ( + statement.level == 0 + and statement.module == "subprocess" + and imported.name == "Popen" + and bound == "Popen" + ): + trusted_names.add(bound) + elif bound in trusted_names: + trusted_names.discard(bound) + + +class _DirectBindingCollector: + """Collect direct receiver bindings without entering nested scopes.""" + + def __init__(self, tracked_names: set[str] | frozenset[str]) -> None: + self.tracked_names = tracked_names + self.bound: set[str] = set() + self.mutated: set[str] = set() + self.nonlocal_names: set[str] = set() + + @staticmethod + def _function_header_nodes( + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> list[ast.AST]: + nodes: list[ast.AST] = [*node.decorator_list, *node.args.defaults] + nodes.extend(item for item in node.args.kw_defaults if item is not None) + arguments = (*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs) + nodes.extend( + argument.annotation for argument in arguments if argument.annotation is not None + ) + if node.args.vararg is not None and node.args.vararg.annotation is not None: + nodes.append(node.args.vararg.annotation) + if node.args.kwarg is not None and node.args.kwarg.annotation is not None: + nodes.append(node.args.kwarg.annotation) + if node.returns is not None: + nodes.append(node.returns) + nodes.extend(getattr(node, "type_params", ())) + return nodes + + def visit(self, node: ast.AST) -> None: + pending = [node] + while pending: + current = pending.pop() + if isinstance(current, ast.Name): + if ( + isinstance(current.ctx, (ast.Store, ast.Del)) + and current.id in self.tracked_names + ): + self.bound.add(current.id) + continue + if isinstance(current, (ast.Attribute, ast.Subscript)): + if isinstance(current.ctx, (ast.Store, ast.Del)): + root: ast.expr = current.value + while isinstance(root, (ast.Attribute, ast.Subscript)): + root = root.value + if isinstance(root, ast.Name) and root.id in self.tracked_names: + self.mutated.add(root.id) + pending.extend(ast.iter_child_nodes(current)) + continue + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef)): + if current.name in self.tracked_names: + self.bound.add(current.name) + pending.extend(self._function_header_nodes(current)) + continue + if isinstance(current, ast.ClassDef): + if current.name in self.tracked_names: + self.bound.add(current.name) + pending.extend(current.decorator_list) + pending.extend(current.bases) + pending.extend(keyword.value for keyword in current.keywords) + continue + if isinstance(current, ast.Lambda): + pending.extend(current.args.defaults) + pending.extend(item for item in current.args.kw_defaults if item is not None) + continue + if isinstance(current, (ast.ListComp, ast.SetComp, ast.GeneratorExp)): + pending.append(current.elt) + for generator in current.generators: + pending.append(generator.iter) + pending.extend(generator.ifs) + continue + if isinstance(current, ast.DictComp): + pending.extend((current.key, current.value)) + for generator in current.generators: + pending.append(generator.iter) + pending.extend(generator.ifs) + continue + if isinstance(current, ast.Import): + for imported in current.names: + bound = imported.asname or imported.name.partition(".")[0] + if bound in self.tracked_names: + self.bound.add(bound) + continue + if isinstance(current, ast.ImportFrom): + if any(imported.name == "*" for imported in current.names): + self.bound.update(self.tracked_names) + continue + for imported in current.names: + bound = imported.asname or imported.name + if bound in self.tracked_names: + self.bound.add(bound) + continue + if isinstance(current, ast.ExceptHandler): + if isinstance(current.name, str) and current.name in self.tracked_names: + self.bound.add(current.name) + pending.extend(ast.iter_child_nodes(current)) + continue + if isinstance(current, (ast.Global, ast.Nonlocal)): + self.nonlocal_names.update(current.names) + continue + if isinstance(current, ast.MatchAs): + if isinstance(current.name, str) and current.name in self.tracked_names: + self.bound.add(current.name) + if current.pattern is not None: + pending.append(current.pattern) + continue + if isinstance(current, ast.MatchStar): + if isinstance(current.name, str) and current.name in self.tracked_names: + self.bound.add(current.name) + continue + if isinstance(current, ast.MatchMapping): + if isinstance(current.rest, str) and current.rest in self.tracked_names: + self.bound.add(current.rest) + pending.extend(current.patterns) + continue + pending.extend(ast.iter_child_nodes(current)) + + +def _function_bound_direct_names( + statement: ast.FunctionDef | ast.AsyncFunctionDef, + tracked_names: set[str], +) -> set[str]: + """Return compile-time local receiver names for one function scope.""" + arguments = statement.args + named = (*arguments.posonlyargs, *arguments.args, *arguments.kwonlyargs) + names = {argument.arg for argument in named} + if arguments.vararg is not None: + names.add(arguments.vararg.arg) + if arguments.kwarg is not None: + names.add(arguments.kwarg.arg) + collector = _DirectBindingCollector(tracked_names) + for child in statement.body: + collector.visit(child) + return names.intersection(tracked_names).union( + collector.bound.difference(collector.nonlocal_names) + ) + + +def _changed_direct_names(nodes: list[ast.AST], tracked_names: set[str]) -> set[str]: + """Return receiver names explicitly rebound or mutated by current-scope nodes.""" + collector = _DirectBindingCollector(tracked_names) + for node in nodes: + collector.visit(node) + return collector.bound.union(collector.mutated) + + +def _class_body_changed_direct_names( + statement: ast.ClassDef, + tracked_names: set[str], +) -> set[str]: + """Return explicit class-execution effects on outer receiver objects.""" + + def nested_classes(node: ast.AST) -> list[ast.ClassDef]: + classes: list[ast.ClassDef] = [] + pending = [node] + while pending: + current = pending.pop() + if isinstance(current, ast.ClassDef): + classes.append(current) + continue + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): + continue + pending.extend(ast.iter_child_nodes(current)) + return classes + + affected: set[str] = set() + pending_classes = [statement] + while pending_classes: + current_class = pending_classes.pop() + declaration_collector = _DirectBindingCollector(tracked_names) + for child in current_class.body: + declaration_collector.visit(child) + global_names = declaration_collector.nonlocal_names.intersection(tracked_names) + + local_direct: dict[str, bool] = {} + affected.update(global_names.intersection(declaration_collector.bound)) + for child in current_class.body: + collector = _DirectBindingCollector(tracked_names) + collector.visit(child) + affected.update( + name + for name in collector.mutated + if name in global_names or local_direct.get(name, True) + ) + affected.update(collector.bound.intersection(global_names)) + pending_classes.extend(nested_classes(child)) + + local_bound = collector.bound.difference(global_names) + if isinstance(child, ast.Import): + for imported in child.names: + bound = imported.asname or imported.name.partition(".")[0] + if bound in local_bound: + local_direct[bound] = ( + imported.name == "subprocess" and bound == "subprocess" + ) + elif isinstance(child, ast.ImportFrom): + for imported in child.names: + bound = imported.asname or imported.name + if bound in local_bound: + local_direct[bound] = ( + child.level == 0 + and child.module == "subprocess" + and imported.name == "Popen" + and bound == "Popen" + ) + elif isinstance(child, ast.Assign): + prior_local_direct = dict(local_direct) + for name in local_bound: + local_direct[name] = False + for target in child.targets: + if isinstance(target, ast.Name) and target.id in local_bound: + local_direct[target.id] = ( + isinstance(child.value, ast.Name) + and child.value.id == target.id + and prior_local_direct.get( + child.value.id, + child.value.id in tracked_names, + ) + ) + elif isinstance(child, ast.AnnAssign) and child.value is not None: + prior_local_direct = dict(local_direct) + for name in local_bound: + local_direct[name] = False + if isinstance(child.target, ast.Name) and child.target.id in local_bound: + local_direct[child.target.id] = ( + isinstance(child.value, ast.Name) + and child.value.id == child.target.id + and prior_local_direct.get( + child.value.id, + child.value.id in tracked_names, + ) + ) + elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + if child.name in tracked_names and child.name not in global_names: + local_direct[child.name] = False + elif isinstance(child, ast.Delete): + for name in collector.bound: + local_direct.pop(name, None) + elif local_bound: + for name in local_bound: + local_direct.pop(name, None) + return affected + + +def _is_direct_subprocess_call(call: ast.Call, trusted_names: set[str]) -> bool: + function = call.func + if isinstance(function, ast.Name): + return function.id == "Popen" and function.id in trusted_names + return ( + isinstance(function, ast.Attribute) + and isinstance(function.value, ast.Name) + and function.value.id == "subprocess" + and function.value.id in trusted_names + ) + + +def _is_passive_argument(expression: ast.expr) -> bool: + """Return whether evaluation cannot invoke user-controlled Python code.""" + normal, hash_required, truth_required, numeric_required, integral_required = range(5) + pending: list[tuple[ast.expr, int]] = [(expression, normal)] + while pending: + current, requirement = pending.pop() + if isinstance(current, ast.Constant): + if requirement == numeric_required and type(current.value) not in ( + bool, + int, + float, + complex, + ): + return False + if requirement == integral_required and type(current.value) not in (bool, int): + return False + continue + if isinstance(current, ast.Name): + if requirement != normal: + return False + continue + if isinstance(current, ast.List): + if requirement in (hash_required, numeric_required, integral_required) or any( + isinstance(item, ast.Starred) for item in current.elts + ): + return False + pending.extend((item, normal) for item in current.elts) + continue + if isinstance(current, ast.Tuple): + if requirement in (numeric_required, integral_required): + return False + if any(isinstance(item, ast.Starred) for item in current.elts): + return False + nested_requirement = hash_required if requirement == hash_required else normal + pending.extend((item, nested_requirement) for item in current.elts) + continue + if isinstance(current, ast.Dict): + if requirement in (hash_required, numeric_required, integral_required) or any( + key is None for key in current.keys + ): + return False + pending.extend((key, hash_required) for key in current.keys if key is not None) + pending.extend((value, normal) for value in current.values) + continue + if isinstance(current, ast.Set): + if requirement in (hash_required, numeric_required, integral_required): + return False + pending.extend((item, hash_required) for item in current.elts) + continue + if isinstance(current, ast.UnaryOp): + if isinstance(current.op, ast.Not): + pending.append((current.operand, truth_required)) + elif isinstance(current.op, (ast.UAdd, ast.USub)): + operand_requirement = ( + integral_required if requirement == integral_required else numeric_required + ) + pending.append((current.operand, operand_requirement)) + elif isinstance(current.op, ast.Invert): + pending.append((current.operand, integral_required)) + else: + return False + continue + if isinstance(current, ast.JoinedStr) and all( + isinstance(item, ast.Constant) for item in current.values + ): + if requirement in (numeric_required, integral_required): + return False + continue + return False + return True + + +def _call_arguments_are_passive(call: ast.Call) -> bool: + return all(_is_passive_argument(argument) for argument in call.args) and all( + keyword.arg is not None and _is_passive_argument(keyword.value) for keyword in call.keywords + ) + + +def _annotation_is_passive(annotation: ast.expr) -> bool: + """Accept only annotation spellings whose evaluation cannot rebind a name.""" + return all( + isinstance(node, (ast.Name, ast.Constant, ast.Load)) for node in ast.walk(annotation) + ) + + +def _function_header_is_passive( + statement: ast.FunctionDef | ast.AsyncFunctionDef, +) -> bool: + """Reject definition-time expressions that could mutate tracked bindings.""" + if statement.decorator_list or getattr(statement, "type_params", []): + return False + defaults = (*statement.args.defaults, *(item for item in statement.args.kw_defaults if item)) + if any(not _is_passive_argument(default) for default in defaults): + return False + arguments = ( + *statement.args.posonlyargs, + *statement.args.args, + *statement.args.kwonlyargs, + ) + annotations = [argument.annotation for argument in arguments if argument.annotation is not None] + if statement.args.vararg is not None and statement.args.vararg.annotation is not None: + annotations.append(statement.args.vararg.annotation) + if statement.args.kwarg is not None and statement.args.kwarg.annotation is not None: + annotations.append(statement.args.kwarg.annotation) + if statement.returns is not None: + annotations.append(statement.returns) + return all(_annotation_is_passive(annotation) for annotation in annotations) + + +def _advance_trusted_names(statement: ast.stmt, trusted_names: set[str]) -> None: + """Apply one statement's explicit receiver-binding effects.""" + if isinstance(statement, (ast.Import, ast.ImportFrom)): + _update_trusted_names_from_import(statement, trusted_names) + return + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + if not _function_header_is_passive(statement): + trusted_names.clear() + trusted_names.discard(statement.name) + return + if isinstance(statement, ast.Assign): + changed = _changed_direct_names( + [statement.value, *statement.targets], + trusted_names, + ) + preserved = { + target.id + for target in statement.targets + if isinstance(target, ast.Name) + and isinstance(statement.value, ast.Name) + and statement.value.id == target.id + and target.id in trusted_names + } + trusted_names.difference_update(changed.difference(preserved)) + return + if isinstance(statement, ast.ClassDef): + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + trusted_names.difference_update(_class_body_changed_direct_names(statement, trusted_names)) + return + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + + +class _Analyzer: + def __init__(self, file_path: str, lines: list[str]) -> None: + self.file_path = file_path + self.lines = lines + self.findings: list[AnalyzerFinding] = [] + + def _inspect_call(self, call: ast.Call, facts: dict[str, bool]) -> None: + shell = next((item.value for item in call.keywords if item.arg == "shell"), None) + if ( + not isinstance(shell, ast.Name) + or shell.id.casefold().startswith("true") + or facts.get(shell.id) is not True + ): + return + line = getattr(call, "lineno", 1) + end_line = getattr(call, "end_lineno", None) + self.findings.append( + AnalyzerFinding( + rule_id="TM1", + message="Tool Parameter Abuse", + severity=Severity.HIGH, + location=Location(file=self.file_path, start_line=line, end_line=end_line), + confidence=0.8, + tags=[PatternCategory.TOOL_MISUSE.value], + context=get_context_from_lines(self.lines, line), + matched_text=get_source_segment(self.lines, line, end_line), + evidence={BOUND_SHELL_EVIDENCE: True}, + ) + ) + + def _scan_assignment( + self, + targets: list[ast.expr], + value: ast.expr, + facts: dict[str, bool], + trusted_names: set[str], + ) -> None: + if isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names): + resolved = None + safe_value = _call_arguments_are_passive(value) + if safe_value: + self._inspect_call(value, facts) + else: + resolved = _truth_value(value, facts) + safe_value = resolved is not None or _is_passive_argument(value) + + if not safe_value or any(not isinstance(target, ast.Name) for target in targets): + facts.clear() + trusted_names.difference_update(_changed_direct_names([value, *targets], trusted_names)) + return + for target in targets: + assert isinstance(target, ast.Name) + if resolved is None: + facts.pop(target.id, None) + else: + facts[target.id] = resolved + preserves_binding = ( + isinstance(value, ast.Name) and value.id == target.id and value.id in trusted_names + ) + if not preserves_binding: + trusted_names.discard(target.id) + + def _scan_block( + self, + statements: list[ast.stmt], + *, + trusted_names: set[str] | None = None, + ) -> None: + trusted_names = set(_DIRECT_CALL_NAMES if trusted_names is None else trusted_names) + facts: dict[str, bool] = {} + last_invalidation_by_name: dict[str, int] = {} + + def last_invalidation(name: str) -> int: + cached = last_invalidation_by_name.get(name) + if cached is not None: + return cached + last = -1 + for candidate_index, candidate in enumerate(statements): + probe = {name} + _advance_trusted_names(candidate, probe) + if name not in probe: + last = candidate_index + last_invalidation_by_name[name] = last + return last + + for index, statement in enumerate(statements): + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + passive_header = _function_header_is_passive(statement) + nested_trusted_names = { + name for name in trusted_names if last_invalidation(name) <= index + } + nested_trusted_names.difference_update( + _function_bound_direct_names(statement, nested_trusted_names) + ) + nested_trusted_names.discard(statement.name) + if not passive_header: + nested_trusted_names.clear() + self._scan_block(statement.body, trusted_names=nested_trusted_names) + if passive_header: + facts.pop(statement.name, None) + else: + facts.clear() + trusted_names.clear() + trusted_names.discard(statement.name) + elif isinstance(statement, (ast.Import, ast.ImportFrom)): + facts.clear() + _update_trusted_names_from_import(statement, trusted_names) + elif isinstance(statement, ast.Assign): + self._scan_assignment( + list(statement.targets), + statement.value, + facts, + trusted_names, + ) + elif isinstance(statement, ast.AnnAssign): + value = statement.value + if ( + isinstance(value, ast.Call) + and _is_direct_subprocess_call(value, trusted_names) + and _call_arguments_are_passive(value) + ): + self._inspect_call(value, facts) + facts.clear() + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + elif isinstance(statement, (ast.AugAssign, ast.Delete)): + facts.clear() + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + elif isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call): + call = statement.value + if _is_direct_subprocess_call(call, trusted_names) and _call_arguments_are_passive( + call + ): + self._inspect_call(call, facts) + else: + facts.clear() + trusted_names.difference_update(_changed_direct_names([call], trusted_names)) + elif isinstance(statement, ast.Pass) or ( + isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant) + ): + continue + elif isinstance(statement, ast.ClassDef): + facts.clear() + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + trusted_names.difference_update( + _class_body_changed_direct_names(statement, trusted_names) + ) + else: + facts.clear() + trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + + def run(self, tree: ast.Module) -> list[AnalyzerFinding]: + self._scan_block(tree.body) + return sorted(self.findings, key=lambda finding: finding.location.start_line) + + +def analyze( + content: str, + file_path: str, + file_type: str, + *, + python_ast: ParsedPythonFile | None = None, +) -> list[AnalyzerFinding]: + """Find straight-line truthy names passed to direct subprocess calls.""" + if file_type != "python": + return [] + parsed = python_ast or parse_python_source(content, file_path) + if parsed.tree is None: + return [] + return _Analyzer(file_path, parsed.lines).run(parsed.tree) From bd09251621e1edabcfb4615b04d420dbf83d02b6 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 20:10:04 -0700 Subject: [PATCH 3/5] fix(analyzer): harden bound shell facts Signed-off-by: Christopher Kevin --- .../static_python_shell_truthiness.py | 251 ++++++++++++++++-- .../analyzers/test_tool_misuse_python_ast.py | 95 ++++++- 2 files changed, 318 insertions(+), 28 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py index 741a9d544..5622c5a55 100644 --- a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py +++ b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py @@ -15,7 +15,7 @@ from skillspector.models import AnalyzerFinding, Location, Severity from skillspector.python_ast import ParsedPythonFile, parse_python_source -from .common import get_context_from_lines, get_source_segment +from .common import get_complete_source_segment, get_context_from_lines from .pattern_defaults import PatternCategory ANALYZER_ID = "static_patterns_tool_misuse" @@ -216,6 +216,56 @@ def visit(self, node: ast.AST) -> None: pending.extend(ast.iter_child_nodes(current)) +def _direct_bound_names(node: ast.AST) -> set[str]: + """Return names bound by *node* without entering deferred nested scopes.""" + candidates: set[str] = set() + for current in ast.walk(node): + if isinstance(current, ast.Name) and isinstance(current.ctx, (ast.Store, ast.Del)): + candidates.add(current.id) + elif isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + candidates.add(current.name) + elif isinstance(current, ast.Import): + candidates.update( + imported.asname or imported.name.partition(".")[0] for imported in current.names + ) + elif isinstance(current, ast.ImportFrom): + candidates.update( + imported.asname or imported.name + for imported in current.names + if imported.name != "*" + ) + elif isinstance(current, ast.ExceptHandler) and isinstance(current.name, str): + candidates.add(current.name) + elif isinstance(current, (ast.MatchAs, ast.MatchStar)) and isinstance( + current.name, + str, + ): + candidates.add(current.name) + elif isinstance(current, ast.MatchMapping) and isinstance(current.rest, str): + candidates.add(current.rest) + collector = _DirectBindingCollector(candidates) + collector.visit(node) + return collector.bound + + +def _function_parameter_names(statement: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]: + """Return names that may already hold unsafe values on body entry.""" + arguments = statement.args + names = { + argument.arg + for argument in (*arguments.posonlyargs, *arguments.args, *arguments.kwonlyargs) + } + if arguments.vararg is not None: + names.add(arguments.vararg.arg) + if arguments.kwarg is not None: + names.add(arguments.kwarg.arg) + declarations = _DirectBindingCollector(set()) + for child in statement.body: + declarations.visit(child) + names.update(declarations.nonlocal_names) + return names + + def _function_bound_direct_names( statement: ast.FunctionDef | ast.AsyncFunctionDef, tracked_names: set[str], @@ -432,6 +482,23 @@ def _call_arguments_are_passive(call: ast.Call) -> bool: ) +def _is_finalizer_safe_value(expression: ast.expr, safe_names: set[str]) -> bool: + """Return whether releasing the resulting value cannot run user code.""" + if not _is_passive_argument(expression): + return False + return all( + not isinstance(node, ast.Name) or node.id in safe_names for node in ast.walk(expression) + ) + + +def _call_arguments_are_protocol_safe(call: ast.Call, safe_names: set[str]) -> bool: + """Return whether subprocess argument consumption cannot dispatch user code.""" + return all(_is_finalizer_safe_value(argument, safe_names) for argument in call.args) and all( + keyword.arg is not None and _is_finalizer_safe_value(keyword.value, safe_names) + for keyword in call.keywords + ) + + def _annotation_is_passive(annotation: ast.expr) -> bool: """Accept only annotation spellings whose evaluation cannot rebind a name.""" return all( @@ -463,6 +530,37 @@ def _function_header_is_passive( return all(_annotation_is_passive(annotation) for annotation in annotations) +def _is_immediate_function(statement: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + """Return whether a direct call begins executing this function body.""" + if isinstance(statement, ast.AsyncFunctionDef): + return False + pending: list[ast.AST] = list(statement.body) + while pending: + current = pending.pop() + if isinstance(current, (ast.Yield, ast.YieldFrom)): + return False + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)): + continue + pending.extend(ast.iter_child_nodes(current)) + return True + + +def _passive_direct_call(statement: ast.stmt) -> ast.Call | None: + """Return a directly evaluated simple-name call with passive arguments.""" + value: ast.expr | None = None + if isinstance(statement, (ast.Expr, ast.Assign)): + value = statement.value + elif isinstance(statement, ast.AnnAssign): + value = statement.value + if ( + isinstance(value, ast.Call) + and isinstance(value.func, ast.Name) + and _call_arguments_are_passive(value) + ): + return value + return None + + def _advance_trusted_names(statement: ast.stmt, trusted_names: set[str]) -> None: """Apply one statement's explicit receiver-binding effects.""" if isinstance(statement, (ast.Import, ast.ImportFrom)): @@ -496,9 +594,10 @@ def _advance_trusted_names(statement: ast.stmt, trusted_names: set[str]) -> None class _Analyzer: - def __init__(self, file_path: str, lines: list[str]) -> None: + def __init__(self, file_path: str, python_ast: ParsedPythonFile) -> None: self.file_path = file_path - self.lines = lines + self.python_ast = python_ast + self.lines = python_ast.lines self.findings: list[AnalyzerFinding] = [] def _inspect_call(self, call: ast.Call, facts: dict[str, bool]) -> None: @@ -511,16 +610,34 @@ def _inspect_call(self, call: ast.Call, facts: dict[str, bool]) -> None: return line = getattr(call, "lineno", 1) end_line = getattr(call, "end_lineno", None) + start_byte_column = getattr(call, "col_offset", 0) + end_byte_column = getattr(call, "end_col_offset", start_byte_column) + start_column = self.python_ast.character_column(line, start_byte_column) + end_column = self.python_ast.character_column(end_line or line, end_byte_column) + complete_match = self.python_ast.source_segment(call) + if complete_match is None: + complete_match = get_complete_source_segment(self.lines, line, end_line) self.findings.append( AnalyzerFinding( rule_id="TM1", message="Tool Parameter Abuse", severity=Severity.HIGH, - location=Location(file=self.file_path, start_line=line, end_line=end_line), + location=Location( + file=self.file_path, + start_line=line, + end_line=end_line, + start_column=start_column, + end_column=end_column, + ), confidence=0.8, tags=[PatternCategory.TOOL_MISUSE.value], - context=get_context_from_lines(self.lines, line), - matched_text=get_source_segment(self.lines, line, end_line), + context=get_context_from_lines( + self.lines, + line, + column=start_column if start_column is not None else 0, + ), + matched_text=complete_match[:200], + complete_match=complete_match, evidence={BOUND_SHELL_EVIDENCE: True}, ) ) @@ -531,26 +648,59 @@ def _scan_assignment( value: ast.expr, facts: dict[str, bool], trusted_names: set[str], + bound_names: set[str], + finalizer_safe_names: set[str], ) -> None: + simple_targets = all(isinstance(target, ast.Name) for target in targets) + releases_unsafe_value = simple_targets and any( + target.id in bound_names + and target.id not in finalizer_safe_names + and not (isinstance(value, ast.Name) and value.id == target.id) + for target in targets + if isinstance(target, ast.Name) + ) + result_is_finalizer_safe = _is_finalizer_safe_value(value, finalizer_safe_names) + call_has_protocol_effects = False if isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names): resolved = None safe_value = _call_arguments_are_passive(value) if safe_value: self._inspect_call(value, facts) + call_has_protocol_effects = not _call_arguments_are_protocol_safe( + value, + finalizer_safe_names, + ) else: resolved = _truth_value(value, facts) safe_value = resolved is not None or _is_passive_argument(value) - if not safe_value or any(not isinstance(target, ast.Name) for target in targets): + if not safe_value or not simple_targets: facts.clear() + finalizer_safe_names.clear() + for target in targets: + if isinstance(target, ast.Name): + bound_names.add(target.id) trusted_names.difference_update(_changed_direct_names([value, *targets], trusted_names)) return + if releases_unsafe_value: + facts.clear() + finalizer_safe_names.clear() + trusted_names.clear() + if call_has_protocol_effects: + facts.clear() + finalizer_safe_names.clear() + trusted_names.clear() for target in targets: assert isinstance(target, ast.Name) - if resolved is None: + bound_names.add(target.id) + if releases_unsafe_value or call_has_protocol_effects or resolved is None: facts.pop(target.id, None) else: facts[target.id] = resolved + if releases_unsafe_value or call_has_protocol_effects or not result_is_finalizer_safe: + finalizer_safe_names.discard(target.id) + else: + finalizer_safe_names.add(target.id) preserves_binding = ( isinstance(value, ast.Name) and value.id == target.id and value.id in trusted_names ) @@ -562,29 +712,52 @@ def _scan_block( statements: list[ast.stmt], *, trusted_names: set[str] | None = None, + initial_bound_names: set[str] | None = None, ) -> None: trusted_names = set(_DIRECT_CALL_NAMES if trusted_names is None else trusted_names) facts: dict[str, bool] = {} - last_invalidation_by_name: dict[str, int] = {} + bound_names = set(initial_bound_names or ()) + finalizer_safe_names: set[str] = set() - def last_invalidation(name: str) -> int: - cached = last_invalidation_by_name.get(name) - if cached is not None: - return cached - last = -1 - for candidate_index, candidate in enumerate(statements): - probe = {name} - _advance_trusted_names(candidate, probe) - if name not in probe: - last = candidate_index - last_invalidation_by_name[name] = last - return last + last_invalidation_by_name: dict[str, int] = {} + receiver_trust = set(trusted_names) + for candidate_index, candidate in enumerate(statements): + before = set(receiver_trust) + _advance_trusted_names(candidate, receiver_trust) + for name in before.difference(receiver_trust): + last_invalidation_by_name[name] = candidate_index + + trusted_at_call_by_definition: dict[int, set[str]] = {} + receiver_trust = set(trusted_names) + active_functions: dict[str, int] = {} + for candidate_index, candidate in enumerate(statements): + call = _passive_direct_call(candidate) + if call is not None: + assert isinstance(call.func, ast.Name) + owner = active_functions.get(call.func.id) + if owner is not None: + trusted_at_call_by_definition.setdefault(owner, set()).update(receiver_trust) + + changed_names = _direct_bound_names(candidate) + for name in changed_names: + active_functions.pop(name, None) + if ( + isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)) + and _function_header_is_passive(candidate) + and _is_immediate_function(candidate) + ): + active_functions[candidate.name] = candidate_index + _advance_trusted_names(candidate, receiver_trust) for index, statement in enumerate(statements): if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): passive_header = _function_header_is_passive(statement) + trusted_at_call = trusted_at_call_by_definition.get(index, set()) + nested_trusted_names = set(trusted_names).union(trusted_at_call) nested_trusted_names = { - name for name in trusted_names if last_invalidation(name) <= index + name + for name in nested_trusted_names + if last_invalidation_by_name.get(name, -1) <= index or name in trusted_at_call } nested_trusted_names.difference_update( _function_bound_direct_names(statement, nested_trusted_names) @@ -592,15 +765,27 @@ def last_invalidation(name: str) -> int: nested_trusted_names.discard(statement.name) if not passive_header: nested_trusted_names.clear() - self._scan_block(statement.body, trusted_names=nested_trusted_names) - if passive_header: + self._scan_block( + statement.body, + trusted_names=nested_trusted_names, + initial_bound_names=_function_parameter_names(statement), + ) + releases_unsafe_value = ( + statement.name in bound_names and statement.name not in finalizer_safe_names + ) + if passive_header and not releases_unsafe_value: facts.pop(statement.name, None) else: facts.clear() + finalizer_safe_names.clear() trusted_names.clear() + bound_names.add(statement.name) + finalizer_safe_names.discard(statement.name) trusted_names.discard(statement.name) elif isinstance(statement, (ast.Import, ast.ImportFrom)): facts.clear() + finalizer_safe_names.clear() + bound_names.update(_direct_bound_names(statement)) _update_trusted_names_from_import(statement, trusted_names) elif isinstance(statement, ast.Assign): self._scan_assignment( @@ -608,6 +793,8 @@ def last_invalidation(name: str) -> int: statement.value, facts, trusted_names, + bound_names, + finalizer_safe_names, ) elif isinstance(statement, ast.AnnAssign): value = statement.value @@ -618,9 +805,14 @@ def last_invalidation(name: str) -> int: ): self._inspect_call(value, facts) facts.clear() + finalizer_safe_names.clear() + if value is not None: + bound_names.update(_direct_bound_names(statement)) trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) elif isinstance(statement, (ast.AugAssign, ast.Delete)): facts.clear() + finalizer_safe_names.clear() + bound_names.update(_direct_bound_names(statement)) trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) elif isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call): call = statement.value @@ -628,8 +820,13 @@ def last_invalidation(name: str) -> int: call ): self._inspect_call(call, facts) + if not _call_arguments_are_protocol_safe(call, finalizer_safe_names): + facts.clear() + finalizer_safe_names.clear() + trusted_names.clear() else: facts.clear() + finalizer_safe_names.clear() trusted_names.difference_update(_changed_direct_names([call], trusted_names)) elif isinstance(statement, ast.Pass) or ( isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant) @@ -637,12 +834,16 @@ def last_invalidation(name: str) -> int: continue elif isinstance(statement, ast.ClassDef): facts.clear() + finalizer_safe_names.clear() + bound_names.update(_direct_bound_names(statement)) trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) trusted_names.difference_update( _class_body_changed_direct_names(statement, trusted_names) ) else: facts.clear() + finalizer_safe_names.clear() + bound_names.update(_direct_bound_names(statement)) trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) def run(self, tree: ast.Module) -> list[AnalyzerFinding]: @@ -663,4 +864,4 @@ def analyze( parsed = python_ast or parse_python_source(content, file_path) if parsed.tree is None: return [] - return _Analyzer(file_path, parsed.lines).run(parsed.tree) + return _Analyzer(file_path, parsed).run(parsed.tree) diff --git a/tests/nodes/analyzers/test_tool_misuse_python_ast.py b/tests/nodes/analyzers/test_tool_misuse_python_ast.py index 64edd9841..47bf9940b 100644 --- a/tests/nodes/analyzers/test_tool_misuse_python_ast.py +++ b/tests/nodes/analyzers/test_tool_misuse_python_ast.py @@ -8,6 +8,7 @@ import pytest from skillspector.nodes.analyzers import static_patterns_tool_misuse as tm_module +from skillspector.nodes.deduplicate import deduplicate def _run(content: str, path: str = "run.py") -> dict: @@ -130,8 +131,8 @@ def test_explicit_import_reestablishes_direct_receivers() -> None: "Popen = Proxy()\n" "from subprocess import Popen\n" "enabled = True\n" - "subprocess.run(command, shell=enabled)\n" - "Popen(command, shell=enabled)\n" + "subprocess.run('/usr/bin/true', shell=enabled)\n" + "Popen('/usr/bin/true', shell=enabled)\n" ) ) == 2 @@ -166,7 +167,19 @@ def test_function_compile_time_receiver_shadow_rejects_earlier_lookup() -> None: ) -def test_later_global_receiver_mutation_suppresses_function_body() -> None: +def test_later_global_receiver_mutation_does_not_suppress_earlier_function_call() -> None: + findings = _tm1( + "def execute(command):\n" + " enabled = True\n" + " subprocess.run(command, shell=enabled)\n" + "execute(command)\n" + "subprocess = Proxy()\n" + ) + + assert [finding.start_line for finding in findings] == [3] + + +def test_later_global_receiver_mutation_suppresses_unobserved_function_body() -> None: assert not _tm1( "def execute(command):\n" " enabled = True\n" @@ -229,6 +242,59 @@ def test_unsupported_assignment_clears_existing_facts() -> None: assert not _tm1("enabled = True\nresult = factory()\nsubprocess.run(cmd, shell=enabled)\n") +def test_simple_name_store_with_unsafe_prior_binding_invalidates_truth_facts() -> None: + findings = _tm1( + "import subprocess\n" + "class Trigger:\n" + " def __del__(self):\n" + " global enabled\n" + " enabled = False\n" + "trigger = Trigger()\n" + "enabled = True\n" + "trigger = 0\n" + "subprocess.run('/usr/bin/true', shell=enabled)\n" + ) + + assert not findings + + +def test_external_name_store_treats_prior_binding_as_finalizer_capable() -> None: + findings = _tm1( + "import subprocess\n" + "class Trigger:\n" + " def __del__(self):\n" + " global enabled\n" + " enabled = False\n" + "trigger = Trigger()\n" + "enabled = False\n" + "def execute():\n" + " global enabled, trigger\n" + " enabled = True\n" + " trigger = 0\n" + " subprocess.run('/usr/bin/true', shell=enabled)\n" + "execute()\n" + ) + + assert not findings + + +def test_protocol_consuming_direct_call_invalidates_later_truth_fact() -> None: + findings = _tm1( + "import subprocess\n" + "class MutatingArgs:\n" + " def __iter__(self):\n" + " global enabled\n" + " enabled = False\n" + " return iter(('/usr/bin/true',))\n" + "mutator = MutatingArgs()\n" + "enabled = True\n" + "subprocess.run(mutator, shell=enabled)\n" + "subprocess.run('/usr/bin/true', shell=enabled)\n" + ) + + assert [finding.start_line for finding in findings] == [9] + + def test_annotated_assignment_is_outside_side_effect_free_contract() -> None: assert not _tm1("enabled: bool = True\nsubprocess.run(command, shell=enabled)\n") @@ -246,6 +312,29 @@ def test_true_prefixed_identifier_has_one_lexical_owner() -> None: assert len(findings) == 1 +def test_long_same_line_calls_keep_exact_coordinates_and_distinct_identity() -> None: + payload = "x" * 240 + first_call = f'subprocess.run("{payload}A", shell=enabled)' + second_call = f'subprocess.run("{payload}B", shell=enabled)' + call_line = f"first = {first_call}; second = {second_call}" + + findings = _tm1(f"import subprocess\nenabled = True\n{call_line}\n") + + assert len(findings) == 2 + assert [(finding.start_column, finding.end_column) for finding in findings] == [ + ( + call_line.index(first_call), + call_line.index(first_call) + len(first_call), + ), + ( + call_line.index(second_call), + call_line.index(second_call) + len(second_call), + ), + ] + assert findings[0].fingerprint() != findings[1].fingerprint() + assert len(deduplicate(findings)) == 2 + + @pytest.mark.parametrize("path", ["run.pyw", "run", "run.sh"]) def test_non_py_surfaces_do_not_enable_ast_companion(path: str) -> None: assert not _tm1("enabled = True\nsubprocess.run(command, shell=enabled)\n", path) From c3b0ff945013f2cc0eeaeb66515ba099d184478a Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Sun, 20 Sep 2026 19:39:02 -0700 Subject: [PATCH 4/5] fix(analyzer): honor shell argument evaluation order Signed-off-by: Christopher Kevin --- .../static_python_shell_truthiness.py | 33 +++++++-- .../analyzers/test_tool_misuse_python_ast.py | 69 +++++++++++++++++++ 2 files changed, 95 insertions(+), 7 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py index 5622c5a55..7b9323e01 100644 --- a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py +++ b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py @@ -4,8 +4,9 @@ """Find direct subprocess calls using a definitely truthy local name. This companion recognizes the straight-line ordinary-Python form reported in -issue #475. Call arguments must be passive, and unsupported expressions or -compound statements discard facts rather than guessing about Python execution. +issue #475. Arguments evaluated through ``shell=`` must be passive, and +unsupported expressions or compound statements discard facts rather than +guessing about Python execution. """ from __future__ import annotations @@ -482,6 +483,23 @@ def _call_arguments_are_passive(call: ast.Call) -> bool: ) +def _shell_argument_is_captured_before_effects(call: ast.Call) -> bool: + """Return whether evaluation reaches ``shell=`` without user-code effects. + + Python evaluates every positional argument, including starred expansions, + before keyword arguments. Keyword values are then evaluated in their stored + order. Effects after ``shell=`` cannot change the already captured value. + """ + if any(not _is_passive_argument(argument) for argument in call.args): + return False + for keyword in call.keywords: + if keyword.arg == "shell": + return _is_passive_argument(keyword.value) + if keyword.arg is None or not _is_passive_argument(keyword.value): + return False + return False + + def _is_finalizer_safe_value(expression: ast.expr, safe_names: set[str]) -> bool: """Return whether releasing the resulting value cannot run user code.""" if not _is_passive_argument(expression): @@ -664,8 +682,9 @@ def _scan_assignment( if isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names): resolved = None safe_value = _call_arguments_are_passive(value) - if safe_value: + if _shell_argument_is_captured_before_effects(value): self._inspect_call(value, facts) + if safe_value: call_has_protocol_effects = not _call_arguments_are_protocol_safe( value, finalizer_safe_names, @@ -801,7 +820,7 @@ def _scan_block( if ( isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names) - and _call_arguments_are_passive(value) + and _shell_argument_is_captured_before_effects(value) ): self._inspect_call(value, facts) facts.clear() @@ -816,10 +835,10 @@ def _scan_block( trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) elif isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call): call = statement.value - if _is_direct_subprocess_call(call, trusted_names) and _call_arguments_are_passive( - call - ): + direct_call = _is_direct_subprocess_call(call, trusted_names) + if direct_call and _shell_argument_is_captured_before_effects(call): self._inspect_call(call, facts) + if direct_call and _call_arguments_are_passive(call): if not _call_arguments_are_protocol_safe(call, finalizer_safe_names): facts.clear() finalizer_safe_names.clear() diff --git a/tests/nodes/analyzers/test_tool_misuse_python_ast.py b/tests/nodes/analyzers/test_tool_misuse_python_ast.py index 47bf9940b..a3fd3d38c 100644 --- a/tests/nodes/analyzers/test_tool_misuse_python_ast.py +++ b/tests/nodes/analyzers/test_tool_misuse_python_ast.py @@ -238,6 +238,75 @@ def test_side_effect_capable_call_arguments_are_rejected(argument: str) -> None: assert not _tm1(f"enabled = True\nsubprocess.run({argument}, shell=enabled)\n") +@pytest.mark.parametrize( + "statement", + [ + pytest.param( + "subprocess.run(command, shell=enabled, env=build_env())", + id="expression", + ), + pytest.param( + "result = subprocess.run(command, shell=enabled, env=build_env())", + id="assignment", + ), + pytest.param( + "result: object = subprocess.run(command, shell=enabled, env=build_env())", + id="annotated-assignment", + ), + ], +) +def test_later_keyword_effect_preserves_captured_shell_value(statement: str) -> None: + findings = _tm1(f"enabled = True\n{statement}\n") + literal_findings = _tm1(statement.replace("shell=enabled", "shell=True")) + + assert len(findings) == len(literal_findings) == 1 + assert findings[0].start_line == 2 + assert findings[0].severity == literal_findings[0].severity + assert findings[0].confidence == literal_findings[0].confidence + + +@pytest.mark.parametrize( + "call", + [ + pytest.param( + "subprocess.run(command, env=build_env(), shell=enabled)", + id="earlier-keyword", + ), + pytest.param( + "subprocess.run(build_command(), shell=enabled)", + id="earlier-positional", + ), + pytest.param( + "subprocess.run(shell=enabled, *build_args())", + id="starred-positional-written-later", + ), + pytest.param( + "subprocess.run(command, **build_options(), shell=enabled)", + id="earlier-keyword-expansion", + ), + ], +) +def test_earlier_argument_effect_keeps_shell_value_uncertain(call: str) -> None: + assert not _tm1(f"enabled = True\n{call}\n") + + +def test_later_keyword_expansion_preserves_captured_shell_value() -> None: + findings = _tm1("enabled = True\nsubprocess.run(command, shell=enabled, **build_options())\n") + + assert len(findings) == 1 + assert findings[0].start_line == 2 + + +def test_later_argument_effect_invalidates_fact_after_captured_call() -> None: + findings = _tm1( + "enabled = True\n" + "subprocess.run(command, shell=enabled, env=build_env())\n" + "subprocess.run(command, shell=enabled)\n" + ) + + assert [finding.start_line for finding in findings] == [2] + + def test_unsupported_assignment_clears_existing_facts() -> None: assert not _tm1("enabled = True\nresult = factory()\nsubprocess.run(cmd, shell=enabled)\n") From 995d746cbc1c615a7bf17c04ed4943c917ce9beb Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Mon, 21 Sep 2026 13:23:47 -0700 Subject: [PATCH 5/5] fix(analyzer): invalidate effectful subprocess receivers Signed-off-by: Christopher Kevin --- .../static_python_shell_truthiness.py | 49 ++++++++++++++----- .../analyzers/test_tool_misuse_python_ast.py | 45 +++++++++++++++++ 2 files changed, 83 insertions(+), 11 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py index 7b9323e01..e0c80d30b 100644 --- a/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py +++ b/src/skillspector/nodes/analyzers/static_python_shell_truthiness.py @@ -580,7 +580,17 @@ def _passive_direct_call(statement: ast.stmt) -> ast.Call | None: def _advance_trusted_names(statement: ast.stmt, trusted_names: set[str]) -> None: - """Apply one statement's explicit receiver-binding effects.""" + """Apply one statement's receiver-trust effects.""" + value = ( + statement.value if isinstance(statement, (ast.Expr, ast.Assign, ast.AnnAssign)) else None + ) + if ( + isinstance(value, ast.Call) + and _is_direct_subprocess_call(value, trusted_names) + and not _call_arguments_are_passive(value) + ): + trusted_names.clear() + return if isinstance(statement, (ast.Import, ast.ImportFrom)): _update_trusted_names_from_import(statement, trusted_names) return @@ -679,9 +689,11 @@ def _scan_assignment( ) result_is_finalizer_safe = _is_finalizer_safe_value(value, finalizer_safe_names) call_has_protocol_effects = False + effectful_direct_call = False if isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names): resolved = None safe_value = _call_arguments_are_passive(value) + effectful_direct_call = not safe_value if _shell_argument_is_captured_before_effects(value): self._inspect_call(value, facts) if safe_value: @@ -699,7 +711,12 @@ def _scan_assignment( for target in targets: if isinstance(target, ast.Name): bound_names.add(target.id) - trusted_names.difference_update(_changed_direct_names([value, *targets], trusted_names)) + if effectful_direct_call: + trusted_names.clear() + else: + trusted_names.difference_update( + _changed_direct_names([value, *targets], trusted_names) + ) return if releases_unsafe_value: facts.clear() @@ -817,17 +834,21 @@ def _scan_block( ) elif isinstance(statement, ast.AnnAssign): value = statement.value - if ( - isinstance(value, ast.Call) - and _is_direct_subprocess_call(value, trusted_names) - and _shell_argument_is_captured_before_effects(value) - ): - self._inspect_call(value, facts) + effectful_direct_call = False + if isinstance(value, ast.Call) and _is_direct_subprocess_call(value, trusted_names): + if _shell_argument_is_captured_before_effects(value): + self._inspect_call(value, facts) + effectful_direct_call = not _call_arguments_are_passive(value) facts.clear() finalizer_safe_names.clear() if value is not None: bound_names.update(_direct_bound_names(statement)) - trusted_names.difference_update(_changed_direct_names([statement], trusted_names)) + if effectful_direct_call: + trusted_names.clear() + else: + trusted_names.difference_update( + _changed_direct_names([statement], trusted_names) + ) elif isinstance(statement, (ast.AugAssign, ast.Delete)): facts.clear() finalizer_safe_names.clear() @@ -838,7 +859,8 @@ def _scan_block( direct_call = _is_direct_subprocess_call(call, trusted_names) if direct_call and _shell_argument_is_captured_before_effects(call): self._inspect_call(call, facts) - if direct_call and _call_arguments_are_passive(call): + arguments_are_passive = _call_arguments_are_passive(call) + if direct_call and arguments_are_passive: if not _call_arguments_are_protocol_safe(call, finalizer_safe_names): facts.clear() finalizer_safe_names.clear() @@ -846,7 +868,12 @@ def _scan_block( else: facts.clear() finalizer_safe_names.clear() - trusted_names.difference_update(_changed_direct_names([call], trusted_names)) + if direct_call: + trusted_names.clear() + else: + trusted_names.difference_update( + _changed_direct_names([call], trusted_names) + ) elif isinstance(statement, ast.Pass) or ( isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant) ): diff --git a/tests/nodes/analyzers/test_tool_misuse_python_ast.py b/tests/nodes/analyzers/test_tool_misuse_python_ast.py index a3fd3d38c..190eb6161 100644 --- a/tests/nodes/analyzers/test_tool_misuse_python_ast.py +++ b/tests/nodes/analyzers/test_tool_misuse_python_ast.py @@ -307,6 +307,51 @@ def test_later_argument_effect_invalidates_fact_after_captured_call() -> None: assert [finding.start_line for finding in findings] == [2] +@pytest.mark.parametrize( + "statement", + [ + pytest.param( + "subprocess.run(command, shell=enabled, env=replace_subprocess())", + id="expression", + ), + pytest.param( + "result = subprocess.run(command, shell=enabled, env=replace_subprocess())", + id="assignment", + ), + pytest.param( + "result: object = subprocess.run(command, shell=enabled, env=replace_subprocess())", + id="annotated-assignment", + ), + ], +) +def test_later_argument_effect_invalidates_receiver_after_captured_call(statement: str) -> None: + findings = _tm1( + "import subprocess\n" + "from helpers import replace_subprocess\n" + "enabled = True\n" + f"{statement}\n" + "later_enabled = True\n" + "subprocess.run(command, shell=later_enabled)\n" + ) + + assert [finding.start_line for finding in findings] == [4] + + +def test_later_argument_effect_invalidates_receiver_for_called_function() -> None: + findings = _tm1( + "import subprocess\n" + "from helpers import replace_subprocess\n" + "enabled = True\n" + "subprocess.run(command, shell=enabled, env=replace_subprocess())\n" + "def execute():\n" + " later_enabled = True\n" + " subprocess.run(command, shell=later_enabled)\n" + "execute()\n" + ) + + assert [finding.start_line for finding in findings] == [4] + + def test_unsupported_assignment_clears_existing_facts() -> None: assert not _tm1("enabled = True\nresult = factory()\nsubprocess.run(cmd, shell=enabled)\n")