fix(tm1): use the final repeated assignment target

src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py

@@ -3811,6 +3811,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
@@ -3830,6 +3902,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):
@@ -3843,16 +3916,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:
@@ -4123,16 +4229,15 @@ 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))
             stack.append((node.target, scope_chain, end))
             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
@@ -4145,7 +4250,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

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