fix(static): detect literal shell flag assignments

src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py

@@ -24,13 +24,15 @@
 
 from __future__ import annotations
 
+import ast
 import re
 import sys
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 
 from skillspector.logging_config import get_logger
 from skillspector.models import AnalyzerFinding, Location, Severity
+from skillspector.python_ast import parse_python_source
 from skillspector.security_reconstruction import validated_json_string_spans
 from skillspector.state import AnalyzerNodeResponse, SkillspectorState
 
@@ -110,11 +112,26 @@
     re.IGNORECASE,
 )
 
+# Match a literal True assigned to a local name shortly before it is passed as
+# shell=<name> to a subprocess invocation.  The bounded newline gap and the
+# intervening-write guard keep this a local data-flow fact; Python scope
+# visibility is enforced separately in analyze() via _VARIABLE_SHELL_FLAG_RE.
+_VARIABLE_SHELL_FLAG_PATTERN = (
+    r"(?m)^\s*([A-Za-z_]\w*)\s*=\s*True\s*$\n"
+    r"(?:(?![^\n]*\b\1\s*=)[^\n]{0,240}\n){0,4}?[^\n]{0,240}"
+    r"(?:subprocess\.\w+|Popen)\s*\([^)]*\bshell\s*=\s*\1\b"
+)
+_VARIABLE_SHELL_FLAG_RE = re.compile(_VARIABLE_SHELL_FLAG_PATTERN, re.IGNORECASE | re.MULTILINE)
+
 # TM1: Tool Parameter Abuse — dangerous parameter values
 TM1_CODE_PATTERNS = [
     # shell=True is a classic command injection vector
     (r"subprocess\.\w+\s*\([^)]*shell\s*=\s*True", 0.8),
     (r"Popen\s*\([^)]*shell\s*=\s*True", 0.8),
+    # Preserve the direct-call signal when a Python boolean is assigned to a
+    # local name immediately before the invocation. The bounded newline gap
+    # avoids treating an arbitrary distant assignment as a data-flow fact.
+    (_VARIABLE_SHELL_FLAG_PATTERN, 0.8),
     # Bound command names on both sides so prefixes such as rmm/ (RAPIDS
     # Memory Manager headers) are not interpreted as destructive commands.
     # Keep the scan within one bounded shell command.  The former ``[^|]*``
@@ -2109,6 +2126,171 @@ def _has_unsupported_brace_expansion(tokens: tuple[_ShellToken, ...]) -> bool:
     )
 
 
+_SCOPE_NODE_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)
+
+
+def _scope_chain(tree: ast.Module, target: ast.AST) -> tuple[ast.AST, ...] | None:
+    """Return the chain of enclosing scopes for *target*, outermost first.
+
+    The module itself is the outermost scope.  Returns None when *target* is
+    not part of *tree*.
+    """
+    parents: dict[ast.AST, ast.AST] = {}
+    stack: list[ast.AST] = [tree]
+    found = False
+    while stack:
+        node = stack.pop()
+        if node is target:
+            found = True
+            break
+        for child in ast.iter_child_nodes(node):
+            parents[child] = node
+            stack.append(child)
+    if not found:
+        return None
+    chain: list[ast.AST] = []
+    current: ast.AST | None = target
+    while current is not None:
+        if isinstance(current, _SCOPE_NODE_TYPES) or current is tree:
+            chain.append(current)
+        current = parents.get(current)
+    chain.reverse()
+    return tuple(chain)
+
+
+def _scope_binds_name(scope_node: ast.AST, name: str) -> bool:
+    """Return whether *name* is bound directly in *scope_node*.
+
+    Nested function/class/lambda bodies are not descended into: their bindings
+    belong to those scopes, not this one.
+    """
+    if isinstance(scope_node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
+        args = scope_node.args
+        named = [arg.arg for arg in (*args.posonlyargs, *args.args, *args.kwonlyargs)]
+        if args.vararg is not None:
+            named.append(args.vararg.arg)
+        if args.kwarg is not None:
+            named.append(args.kwarg.arg)
+        if name in named:
+            return True
+        bodies: list[ast.AST] = (
+            [scope_node.body] if isinstance(scope_node, ast.Lambda) else list(scope_node.body)
+        )
+    elif isinstance(scope_node, (ast.ClassDef, ast.Module)):
+        bodies = list(scope_node.body)
+    else:
+        return False
+    stack = list(bodies)
+    while stack:
+        node = stack.pop()
+        if isinstance(node, _SCOPE_NODE_TYPES):
+            continue
+        if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store) and node.id == name:
+            return True
+        stack.extend(ast.iter_child_nodes(node))
+    return False
+
+
+def _direct_global_nonlocal(scope_node: ast.AST, name: str) -> str | None:
+    """Return 'global'/'nonlocal' when *scope_node* declares *name* as such.
+
+    Only declarations directly in the scope are considered; nested scopes are
+    not descended into.
+    """
+    if isinstance(scope_node, ast.Lambda):
+        return None
+    bodies: list[ast.stmt] | None = None
+    if isinstance(scope_node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module)):
+        bodies = scope_node.body
+    if bodies is None:
+        return None
+    stack: list[ast.AST] = list(bodies)
+    while stack:
+        node = stack.pop()
+        if isinstance(node, _SCOPE_NODE_TYPES):
+            continue
+        if isinstance(node, ast.Global) and name in node.names:
+            return "global"
+        if isinstance(node, ast.Nonlocal) and name in node.names:
+            return "nonlocal"
+        stack.extend(ast.iter_child_nodes(node))
+    return None
+
+
+def _variable_shell_flag_same_scope(content: str, file_path: str, match: re.Match[str]) -> bool:
+    """Return whether a variable-shell-flag match is a same-scope data flow.
+
+    The regex cannot see Python scopes, so ``use_shell = True`` in one
+    function followed by ``shell=use_shell`` in another still matches.  Resolve
+    the matched assignment and the ``shell=`` use through the Python AST and
+    require the assignment to be visible from the use: identical scope chains,
+    a closure read from an enclosing scope, or a matching global/nonlocal
+    declaration.  Unparseable content keeps the candidate so a syntax error
+    cannot silence the signal.
+    """
+    var_name = match.group(1)
+    assign_line = content.count("\n", 0, match.start()) + 1
+    use_line = content.count("\n", 0, match.end()) + 1
+    tree = parse_python_source(content, file_path).tree
+    if tree is None:
+        return True
+    assign_node: ast.Assign | None = None
+    call_node: ast.Call | None = None
+    for node in ast.walk(tree):
+        if (
+            assign_node is None
+            and isinstance(node, ast.Assign)
+            and node.lineno == assign_line
+            and isinstance(node.value, ast.Constant)
+            and node.value.value is True
+            and any(
+                isinstance(target, ast.Name) and target.id == var_name for target in node.targets
+            )
+        ):
+            assign_node = node
+        if (
+            call_node is None
+            and isinstance(node, ast.Call)
+            and any(
+                keyword.arg == "shell"
+                and isinstance(keyword.value, ast.Name)
+                and keyword.value.id == var_name
+                and assign_line <= keyword.value.lineno <= use_line
+                for keyword in node.keywords
+            )
+        ):
+            call_node = node
+    if assign_node is None or call_node is None:
+        return True
+    assign_chain = _scope_chain(tree, assign_node)
+    use_chain = _scope_chain(tree, call_node)
+    if assign_chain is None or use_chain is None:
+        return True
+    if assign_chain == use_chain:
+        return True
+    if len(assign_chain) < len(use_chain) and use_chain[: len(assign_chain)] == assign_chain:
+        # Closure read: the use sits in a scope nested inside the assignment's
+        # scope, so the name resolves to the assigned value.
+        return True
+    use_scope = use_chain[-1]
+    if use_scope is not tree:
+        declaration = _direct_global_nonlocal(use_scope, var_name)
+        if declaration == "global":
+            return assign_chain == (tree,)
+        if declaration == "nonlocal":
+            binding = next(
+                (
+                    scope
+                    for scope in use_chain[-2::-1]
+                    if isinstance(scope, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda))
+                    and _scope_binds_name(scope, var_name)
+                ),
+                None,
+            )
+            return binding is not None and assign_chain == _scope_chain(tree, binding)
+    return False
+
+
 def _tm1_candidates(
     content: str,
 ) -> Iterator[tuple[int, int, str, float]]:
@@ -2577,7 +2759,18 @@ def ctx(start: int) -> str:
     tag = [PatternCategory.TOOL_MISUSE.value]
     tm1_findings_by_key: dict[tuple[int, str], AnalyzerFinding] = {}
 
+    # The variable-shell-flag regex cannot see Python scopes, so an assignment
+    # in one function and a shell= use in another still match.  Drop those
+    # cross-scope candidates for Python files.
+    cross_scope_starts: set[int] = set()
+    if file_type == "python":
+        for variable_match in _VARIABLE_SHELL_FLAG_RE.finditer(content):
+            if not _variable_shell_flag_same_scope(content, file_path, variable_match):
+                cross_scope_starts.add(variable_match.start())
+
     for match_start, match_end, matched_text, confidence in _tm1_candidates(content):
+        if match_start in cross_scope_starts:
+            continue
         line_num = get_line_number(content, match_start)
         context_text = ctx(match_start)
         matched = matched_text[:200]

tests/unit/test_patterns_new.py

@@ -1448,6 +1448,14 @@ class TestToolMisuse:
         [
             pytest.param("subprocess.run(cmd, shell=True)", "runner.py", "python", id="shell_true"),
             pytest.param("Popen(cmd, shell=True)", "runner.py", "python", id="popen_shell_true"),
+            pytest.param(
+                "command = 'python a.py'\n"
+                "use_shell = True\n"
+                "subprocess.run(command, shell=use_shell)",
+                "runner.py",
+                "python",
+                id="static_true_shell_variable",
+            ),
             pytest.param("rm -rf /", "cleanup.sh", "shell", id="rm_rf_root"),
             pytest.param("chmod 777 /tmp/secrets", "setup.sh", "shell", id="chmod_777"),
             pytest.param("git push --force", "deploy.sh", "shell", id="git_force_push"),
@@ -1465,6 +1473,45 @@ def test_tm1_confidence_boost_for_python(self) -> None:
         tm1 = [f for f in findings if f.rule_id == "TM1"]
         assert all(f.confidence >= 0.8 for f in tm1)
 
+    def test_tm1_ignores_reassigned_shell_variable(self) -> None:
+        findings = tm_mod.analyze(
+            "use_shell = True\nuse_shell = False\nsubprocess.run(cmd, shell=use_shell)",
+            "runner.py",
+            "python",
+        )
+        assert not any(finding.rule_id == "TM1" for finding in findings)
+
+    def test_tm1_ignores_shell_variable_from_unrelated_scope(self) -> None:
+        content = (
+            "def helper():\n"
+            "    use_shell = True\n"
+            "\n"
+            "def main():\n"
+            "    subprocess.run(cmd, shell=use_shell)\n"
+        )
+        findings = tm_mod.analyze(content, "runner.py", "python")
+        assert not any(finding.rule_id == "TM1" for finding in findings)
+
+    def test_tm1_keeps_shell_variable_read_through_closure(self) -> None:
+        content = (
+            "def outer():\n"
+            "    use_shell = True\n"
+            "    def inner():\n"
+            "        subprocess.run(cmd, shell=use_shell)\n"
+        )
+        findings = tm_mod.analyze(content, "runner.py", "python")
+        assert any(finding.rule_id == "TM1" for finding in findings)
+
+    def test_tm1_keeps_global_shell_variable(self) -> None:
+        content = (
+            "use_shell = True\n"
+            "def main():\n"
+            "    global use_shell\n"
+            "    subprocess.run(cmd, shell=use_shell)\n"
+        )
+        findings = tm_mod.analyze(content, "runner.py", "python")
+        assert any(finding.rule_id == "TM1" for finding in findings)
+
     def test_application_specific_no_verify_flag_is_not_tool_misuse(self) -> None:
         content = """\
 print("verification: skipped (--no-verify)")