From 5199a2ffb22dccc77a5779890f30d9b5acd08c3b Mon Sep 17 00:00:00 2001 From: Deepak Jain Date: Wed, 16 Sep 2026 09:03:47 -0700 Subject: [PATCH 1/3] fix(static): detect literal shell flag assignments Signed-off-by: Deepak Jain --- .../nodes/analyzers/static_patterns_tool_misuse.py | 9 +++++++++ tests/unit/test_patterns_new.py | 8 ++++++++ 2 files changed, 17 insertions(+) diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index 0ca9f753c..34f9ff592 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -115,6 +115,15 @@ # 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. + ( + r"(?m)^\s*([A-Za-z_]\w*)\s*=\s*True\s*$\n" + r"(?:[^\n]{0,240}\n){0,4}?[^\n]{0,240}" + r"(?:subprocess\.\w+|Popen)\s*\([^)]*\bshell\s*=\s*\1\b", + 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 ``[^|]*`` diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index 576d1ca64..21d8b0a80 100644 --- a/tests/unit/test_patterns_new.py +++ b/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"), From 426b0ed947941e9e42b6647c2467e377ce21ef8a Mon Sep 17 00:00:00 2001 From: Deepak Jain Date: Wed, 16 Sep 2026 14:48:31 -0700 Subject: [PATCH 2/3] fix(static): respect shell flag reassignment Signed-off-by: Deepak Jain --- .../nodes/analyzers/static_patterns_tool_misuse.py | 2 +- tests/unit/test_patterns_new.py | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index 34f9ff592..0ce9b407c 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -120,7 +120,7 @@ # avoids treating an arbitrary distant assignment as a data-flow fact. ( r"(?m)^\s*([A-Za-z_]\w*)\s*=\s*True\s*$\n" - r"(?:[^\n]{0,240}\n){0,4}?[^\n]{0,240}" + r"(?:(?![^\n]*\b\1\s*=)[^\n]{0,240}\n){0,4}?[^\n]{0,240}" r"(?:subprocess\.\w+|Popen)\s*\([^)]*\bshell\s*=\s*\1\b", 0.8, ), diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index 21d8b0a80..c212437b0 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -1473,6 +1473,14 @@ 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_application_specific_no_verify_flag_is_not_tool_misuse(self) -> None: content = """\ print("verification: skipped (--no-verify)") From d830a7108466515ab21df8b141bf2a5fe0e9d83c Mon Sep 17 00:00:00 2001 From: Deepak Jain Date: Sun, 20 Sep 2026 05:39:51 +0000 Subject: [PATCH 3/3] fix(static): require same-scope shell flag assignment The variable shell flag regex matched across Python scopes, treating use_shell = True in one function and shell=use_shell in another as one data-flow fact. Resolve the matched assignment and use through the Python AST and drop cross-scope candidates, while keeping closure reads and global/nonlocal declarations. Adds the unrelated-scope negative regression. Signed-off-by: Deepak Jain --- .../analyzers/static_patterns_tool_misuse.py | 196 +++++++++++++++++- tests/unit/test_patterns_new.py | 31 +++ 2 files changed, 221 insertions(+), 6 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index 0ce9b407c..58ae95326 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -24,6 +24,7 @@ from __future__ import annotations +import ast import re import sys from collections.abc import Callable, Iterator @@ -31,6 +32,7 @@ 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,6 +112,17 @@ re.IGNORECASE, ) +# Match a literal True assigned to a local name shortly before it is passed as +# shell= 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 @@ -118,12 +131,7 @@ # 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. - ( - 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", - 0.8, - ), + (_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 ``[^|]*`` @@ -2118,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]]: @@ -2586,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] diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index c212437b0..50c410b2a 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -1481,6 +1481,37 @@ def test_tm1_ignores_reassigned_shell_variable(self) -> None: ) 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)")