From f60b3a6334d2be3cd97ecdafba37c195f0b13d7e Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 14 Oct 2026 08:58:56 -0700 Subject: [PATCH 1/2] fix(tm1): preserve final repeated assignment state Signed-off-by: Christopher Kevin --- .../analyzers/static_patterns_tool_misuse.py | 123 ++++++++++++++++-- 1 file changed, 114 insertions(+), 9 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index a0ef03d94..238c3bfa6 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -3691,6 +3691,78 @@ def _direct_shell_name(call: ast.Call) -> str | None: return None +def _passive_assignment_name_snapshots(tree: ast.AST) -> dict[ast.AST, frozenset[str]]: + """Prove prior simple-name bindings in supported straight-line scopes.""" + from . import static_python_shell_truthiness + + snapshots: dict[ast.AST, frozenset[str]] = {} + for scope in ast.walk(tree): + if not isinstance(scope, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef)): + continue + bound_names = ( + set(_parameter_names(scope.args)) if not isinstance(scope, ast.Module) else set() + ) + declarations = static_python_shell_truthiness._DirectBindingCollector(frozenset()) + for statement in scope.body: + declarations.visit(statement) + bound_names.update(declarations.nonlocal_names) + safe_names: set[str] = set() + trusted_names = set(static_python_shell_truthiness._DIRECT_CALL_NAMES).difference( + bound_names + ) + unknown_effects = [False] + for statement in scope.body: + if isinstance(statement, (ast.Assign, ast.AnnAssign)) and not unknown_effects[0]: + targets = ( + statement.targets if isinstance(statement, ast.Assign) else [statement.target] + ) + value = statement.value + if value is not None: + pairs = [ + pair for target in targets for pair in _paired_store_values(target, value) + ] + if pairs and all( + isinstance(target, ast.Name) + and (target.id not in bound_names or target.id in safe_names) + for target, _ in pairs + ): + snapshots[statement] = frozenset( + item.id + for item in ast.walk(value) + if isinstance(item, ast.Name) and item.id in safe_names + ) + if isinstance(statement, (ast.Import, ast.ImportFrom)): + native_import = ( + isinstance(statement, ast.Import) + and all( + alias.name == "subprocess" and alias.asname in (None, "subprocess") + for alias in statement.names + ) + or isinstance(statement, ast.ImportFrom) + and statement.level == 0 + and statement.module == "subprocess" + and all( + alias.name == "Popen" and alias.asname in (None, "Popen") + for alias in statement.names + ) + ) + if not native_import: + # Arbitrary module initialization can change existing names. + unknown_effects[0] = True + safe_names.clear() + static_python_shell_truthiness._advance_trusted_names( + statement, trusted_names, bound_names, safe_names, unknown_effects + ) + if isinstance(statement, (ast.Import, ast.ImportFrom)) and not unknown_effects[0]: + # Only a fresh, proven native import adds release-safe names. + safe_names.update( + trusted_names.intersection( + static_python_shell_truthiness._direct_bound_names(statement) + ) + ) + return snapshots + + def _build_variable_shell_ast_index(parsed: ParsedPythonFile) -> _VariableShellAstIndex | None: """Index relevant nodes, bindings, and declarations in one AST traversal.""" tree = parsed.tree @@ -3710,6 +3782,7 @@ def _build_variable_shell_ast_index(parsed: ParsedPythonFile) -> _VariableShellA # self-stores. They neither replace nor restore a binding. identity_name_stores: set[ast.Name] = set() native_aliases = _native_receiver_aliases(tree) + passive_assignment_snapshots = _passive_assignment_name_snapshots(tree) def mark_attribute_identity_stores(target: ast.AST, value: ast.AST) -> None: for destination, source in _paired_store_values(target, value): @@ -3723,16 +3796,49 @@ def mark_attribute_identity_stores(target: ast.AST, value: ast.AST) -> None: ): attribute_identity_stores.add(destination) - def record_name_store_values(scope: ast.AST, target: ast.AST, value: ast.AST, end: int) -> None: + def record_name_store_values( + scope: ast.AST, targets: list[ast.AST], value: ast.AST, end: int + ) -> None: + from . import static_python_shell_truthiness + # Pair unpacked targets with their RHS elements so a native alias, a # truthy flag, or a self-store is not mistaken for a replacement. - for destination, source in _paired_store_values(target, value): - if not isinstance(destination, ast.Name): - continue + pairs = [ + (destination, source) + for target in targets + for destination, source in reversed(list(_paired_store_values(target, value))) + ] + stores = [ + (destination, source) + for destination, source in pairs + if isinstance(destination, ast.Name) + ] + # The paired iterator is stack-based. Restore Python's left-to-right + # target order before deciding which occurrence owns the final value. + final_values = {destination.id: source for destination, source in stores} + safe_names = passive_assignment_snapshots.get(node) + safe_stores = safe_names is not None + safe_value = ( + safe_names is not None + and static_python_shell_truthiness._is_finalizer_safe_value(value, set(safe_names)) + ) + # Unsupported RHS effects or target protocols cannot prove the final + # flag. Preserve the existing conservative event policy for those shapes. + false_values = ( + { + source + for source in set(final_values.values()) + if static_python_shell_truthiness._truth_value(source, {}) is False + } + if safe_stores and safe_value + else set() + ) + definitely_false = {name for name, source in final_values.items() if source in false_values} + for destination, source in stores: keeps_signal = _store_keeps_shell_signal(destination.id, source, native_aliases) if keeps_signal is None: identity_name_stores.add(destination) - elif keeps_signal: + elif keeps_signal and destination.id not in definitely_false: record_binding(scope, destination.id, end, True) def record_binding(scope: ast.AST, name: str, start: int, restores: bool = False) -> None: @@ -4003,8 +4109,7 @@ def bind(scope: ast.AST, name: str, start: int) -> None: if isinstance(node, ast.Assign): stack.append((node.value, scope_chain, None)) stack.extend((target, scope_chain, end) for target in node.targets) - for target in node.targets: - record_name_store_values(scope, target, node.value, end) + record_name_store_values(scope, list(node.targets), node.value, end) elif isinstance(node, (ast.AnnAssign, ast.AugAssign)): if node.value is not None: stack.append((node.value, scope_chain, None)) @@ -4012,7 +4117,7 @@ def bind(scope: ast.AST, name: str, start: int) -> None: if isinstance(node, ast.AnnAssign): stack.append((node.annotation, scope_chain, None)) if node.value is not None: - record_name_store_values(scope, node.target, node.value, end) + record_name_store_values(scope, [node.target], node.value, end) elif isinstance(node, ast.NamedExpr): stack.append((node.value, scope_chain, None)) # PEP 572 makes a walrus target inside one or more @@ -4025,7 +4130,7 @@ def bind(scope: ast.AST, name: str, start: int) -> None: ): target_chain = target_chain[:-1] stack.append((node.target, target_chain, end)) - record_name_store_values(target_chain[-1], node.target, node.value, end) + record_name_store_values(target_chain[-1], [node.target], node.value, end) elif isinstance(node, (ast.For, ast.AsyncFor)): iteration_span = _node_character_span(parsed, node.iter) iteration_end = iteration_span[1] if iteration_span is not None else end From 0ecb722c6c5de088f8806bb1a8b59311de0aad75 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 14 Oct 2026 08:58:56 -0700 Subject: [PATCH 2/2] test(tm1): cover final assignment state and identity Signed-off-by: Christopher Kevin --- .../test_tm1_assignment_final_state.py | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 tests/nodes/analyzers/test_tm1_assignment_final_state.py diff --git a/tests/nodes/analyzers/test_tm1_assignment_final_state.py b/tests/nodes/analyzers/test_tm1_assignment_final_state.py new file mode 100644 index 000000000..e63871acf --- /dev/null +++ b/tests/nodes/analyzers/test_tm1_assignment_final_state.py @@ -0,0 +1,170 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Repeated boolean targets must use their final assigned value.""" + +from __future__ import annotations + +import pytest + +from skillspector.nodes.analyzers import static_patterns_tool_misuse as tm + + +def _scan(statement: str) -> list: + source = ( + "import subprocess\n" + "enabled = True\n" + statement + "subprocess.run('echo ok', shell=enabled)\n" + ) + result = tm.node({"components": ["runner.py"], "file_cache": {"runner.py": source}}) + return [finding for finding in result["findings"] if finding.rule_id == "TM1"] + + +@pytest.mark.parametrize("false_value", ["False", "0", "''", "None", "()"]) +@pytest.mark.parametrize( + "assignment", + [ + "enabled, enabled = True, {value}\n", + "[enabled, enabled] = [True, {value}]\n", + "enabled, (enabled, enabled) = False, (True, {value})\n", + "enabled = enabled, enabled = True, {value}\n", + ], +) +def test_last_false_store_drops_earlier_truthy_value(assignment: str, false_value: str) -> None: + assert not _scan(assignment.format(value=false_value)) + + +@pytest.mark.parametrize( + "assignment", + [ + "enabled, enabled = False, True\n", + "[enabled, enabled] = [False, True]\n", + "enabled, (enabled, enabled) = False, (False, True)\n", + "enabled = enabled, enabled = False, True\n", + ], +) +def test_last_true_store_keeps_detection(assignment: str) -> None: + findings = _scan(assignment) + assert len(findings) == 1 + assert findings[0].severity == "HIGH" + + +def test_direct_boolean_controls() -> None: + assert len(_scan("")) == 1 + assert not _scan("enabled = False\n") + + +@pytest.mark.parametrize( + "assignment", + [ + "enabled, enabled, enabled = True, False, enabled\n", + "[enabled, enabled, enabled] = [True, False, enabled]\n", + "enabled, (enabled, enabled) = True, (False, enabled)\n", + "enabled = enabled, enabled = True, enabled\n", + "enabled, enabled = True, dynamic\n", + ], +) +def test_other_final_values_preserve_existing_event_policy(assignment: str) -> None: + assert len(_scan(assignment)) == 1 + + +@pytest.mark.parametrize( + "assignment", + [ + "enabled, enabled, enabled = True, opaque, False\n", + "enabled, enabled, enabled = True, factory(), False\n", + "enabled, enabled, other.value = True, False, 0\n", + ], +) +def test_opaque_assignment_effects_preserve_existing_event_policy(assignment: str) -> None: + assert len(_scan(assignment)) == 1 + + +@pytest.mark.parametrize( + "statement", + [ + "other = factory()\nenabled, enabled, other = True, False, None\n", + "enabled = factory()\nenabled, enabled = True, False\n", + ], +) +def test_unknown_prior_values_preserve_existing_event_policy(statement: str) -> None: + assert len(_scan(statement)) == 1 + + +@pytest.mark.parametrize("parameter", [True, False]) +def test_local_target_requires_known_prior_value(parameter: bool) -> None: + source = "import subprocess\ndef execute(" + ("other" if parameter else "") + "):\n" + if not parameter: + source += " other = None\n" + source += ( + " enabled = True\n" + " enabled, enabled, other = True, False, None\n" + " subprocess.run('echo ok', shell=enabled)\n" + ) + response = tm.node({"components": ["runner.py"], "file_cache": {"runner.py": source}}) + hits = [finding for finding in response["findings"] if finding.rule_id == "TM1"] + assert len(hits) == int(parameter) + + +def test_class_scope_preserves_existing_event_policy() -> None: + source = ( + "import subprocess\n" + "class Options:\n" + " enabled = True\n" + " enabled, enabled = True, False\n" + " subprocess.run('echo ok', shell=enabled)\n" + ) + response = tm.node({"components": ["runner.py"], "file_cache": {"runner.py": source}}) + hits = [finding for finding in response["findings"] if finding.rule_id == "TM1"] + assert len(hits) == 1 + + +def test_other_imports_preserve_existing_event_policy() -> None: + assert len(_scan("import math\nenabled, enabled = True, False\n")) == 1 + + +@pytest.mark.parametrize( + "parameter, native_import", + [("Popen", "import subprocess"), ("subprocess", "from subprocess import Popen")], +) +def test_native_import_does_not_make_other_parameter_safe( + parameter: str, native_import: str +) -> None: + source = ( + "import subprocess\n" + f"def execute({parameter}):\n" + f" {native_import}\n" + " enabled = True\n" + f" enabled, enabled, {parameter} = True, False, None\n" + + ( + " subprocess.run('echo ok', shell=enabled)\n" + if parameter == "Popen" + else " Popen('echo ok', shell=enabled)\n" + ) + ) + response = tm.node({"components": ["runner.py"], "file_cache": {"runner.py": source}}) + hits = [finding for finding in response["findings"] if finding.rule_id == "TM1"] + assert len(hits) == 1 + + +@pytest.mark.parametrize( + "target, native_import, receiver", + [ + ("Popen", "from subprocess import Popen", "subprocess.run"), + ("subprocess", "import subprocess", "Popen"), + ], +) +def test_fresh_native_import_keeps_its_own_binding_safe( + target: str, native_import: str, receiver: str +) -> None: + source = ( + "import subprocess\n" + "from subprocess import Popen\n" + "def execute():\n" + f" {native_import}\n" + " enabled = True\n" + f" enabled, enabled, {target} = True, False, None\n" + f" {receiver}('echo ok', shell=enabled)\n" + ) + response = tm.node({"components": ["runner.py"], "file_cache": {"runner.py": source}}) + hits = [finding for finding in response["findings"] if finding.rule_id == "TM1"] + assert not hits