diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index b64592b93..6e9be21e8 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -44,8 +44,8 @@ LINE_BREAK_CHARS, MARKDOWN_FENCE_CLOSE, MARKDOWN_FENCE_OPEN, - get_context, - get_line_number, + SourceLocationIndex, + get_context_from_lines, is_reference_material, ) from .pattern_defaults import PatternCategory @@ -53,6 +53,7 @@ logger = get_logger(__name__) ANALYZER_ID = "static_patterns_tool_misuse" +USES_RUNTIME_CHECK = True ANALYZE_USES_POSTPROCESS = True POSTPROCESS_USES_PYTHON_AST = True _VARIABLE_SHELL_FLAG_EVIDENCE = "_tm1_variable_shell_flag" @@ -4715,16 +4716,22 @@ def analyze( file_path: str, file_type: str, *, + check_runtime: Callable[[], None] | None = None, defer_variable_reconciliation: bool = False, ) -> list[AnalyzerFinding]: """Analyze content for tool misuse patterns (TM1–TM3).""" + runtime_check = check_runtime or (lambda: None) + runtime_check() findings: list[AnalyzerFinding] = [] + locations = SourceLocationIndex(content, file_path) + lines = content.splitlines() def loc(ln: int) -> Location: return Location(file=file_path, start_line=ln) def ctx(start: int) -> str: - return get_context(content, start) + line, column = locations.line_and_column(start) + return get_context_from_lines(lines, line, column=column) tag = [PatternCategory.TOOL_MISUSE.value] tm1_findings_by_key: dict[tuple[int, str, int], AnalyzerFinding] = {} @@ -4735,10 +4742,14 @@ def ctx(start: int) -> str: } invisible_variable_matches: set[tuple[int, int]] = set() if file_type == "python" and variable_matches and not defer_variable_reconciliation: + runtime_check() parsed = parse_python_source(content, file_path) + runtime_check() ast_index = _build_variable_shell_ast_index(parsed) + runtime_check() if ast_index is not None: for span, (_, match) in variable_matches.items(): + runtime_check() assignment_line = bisect_right(ast_index.line_character_starts, match.start(1)) candidate = _resolve_variable_shell_candidate(ast_index, match, assignment_line) if candidate is not None and ( @@ -4750,11 +4761,12 @@ def ctx(start: int) -> str: invisible_variable_matches.add(span) shell_content = ( - _perl_literal_print_shell_text(content, lambda: None) if file_type == "perl" else None + _perl_literal_print_shell_text(content, runtime_check) if file_type == "perl" else None ) for match_start, match_end, matched_text, confidence in _tm1_candidates( content, shell_content=shell_content ): + runtime_check() variable_match = variable_matches.get((match_start, match_end)) if variable_match is not None and variable_match[0].casefold().startswith("true"): # The case-insensitive direct ``shell=True`` pattern already owns @@ -4762,7 +4774,7 @@ def ctx(start: int) -> str: continue if variable_match is not None and (match_start, match_end) in invisible_variable_matches: continue - line_num = get_line_number(content, match_start) + line_num = locations.line_and_column(match_start)[0] context_text = ctx(match_start) matched = matched_text[:200] matched_line = _line_containing(content, match_start, match_end) @@ -4807,7 +4819,8 @@ def ctx(start: int) -> str: else re.finditer ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): - line_num = get_line_number(content, match.start()) + runtime_check() + line_num = locations.line_and_column(match.start())[0] context_text = ctx(match.start()) matched = match.group(0)[:200] @@ -4837,7 +4850,8 @@ def ctx(start: int) -> str: else re.finditer ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): - line_num = get_line_number(content, match.start()) + runtime_check() + line_num = locations.line_and_column(match.start())[0] findings.append( AnalyzerFinding( rule_id="TM3", @@ -4859,7 +4873,8 @@ def ctx(start: int) -> str: tm4_tags = [*tag, "contextual-triage", "likely-benign-context"] if reference_material else tag for pattern, confidence in TM4_PATTERNS: for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE): - line_num = get_line_number(content, match.start()) + runtime_check() + line_num = locations.line_and_column(match.start())[0] findings.append( AnalyzerFinding( rule_id="TM4", diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index c45afdb72..43ea2f134 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -1527,6 +1527,67 @@ def test_tm1_keeps_global_shell_variable(self) -> None: findings = tm_mod.analyze(content, "runner.py", "python") assert any(finding.rule_id == "TM1" for finding in findings) + def test_tm1_parses_dense_shell_flags_once(self) -> None: + content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 4000 + content += "#" + " ordinary" * ((256_000 - len(content)) // 9) + content = content.ljust(256_000) + assert len(content) == 256_000 + with patch.object(tm_mod, "parse_python_source", wraps=tm_mod.parse_python_source) as parse: + findings = tm_mod.analyze(content, "runner.py", "python") + assert parse.call_count == 1 + assert len([finding for finding in findings if finding.rule_id == "TM1"]) == 4000 + + def test_tm1_caches_unparseable_window_conservatively(self) -> None: + content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 20 + "\nif :" + with patch.object(tm_mod, "parse_python_source", wraps=tm_mod.parse_python_source) as parse: + findings = tm_mod.analyze(content, "runner.py", "python") + assert parse.call_count == 1 + assert len([finding for finding in findings if finding.rule_id == "TM1"]) == 20 + + def test_tm1_remains_lexical_for_large_and_normalized_views(self) -> None: + from skillspector.nodes.analyzers import static_runner + + for content in ( + "# ordinary source\n" * 16_000 + "subprocess.run(cmd, shell=True)", + "subprocess.run(cmd, shell=Tru\u200be)", + ): + result = static_runner.run_static_patterns_with_ledger( + {"components": ["runner.py"], "file_cache": {"runner.py": content}}, [tm_mod] + ) + assert any(finding.rule_id == "TM1" for finding in result["findings"]) + + def test_tm1_scope_index_honors_runtime_check(self) -> None: + content = "use_shell = True\nsubprocess.run(cmd, shell=use_shell)\n" * 300 + calls = 0 + + def check_runtime() -> None: + nonlocal calls + calls += 1 + if calls == 20: + raise RuntimeError("deadline") + + with pytest.raises(RuntimeError, match="deadline"): + tm_mod.analyze(content, "runner.py", "python", check_runtime=check_runtime) + assert calls == 20 + + def test_tm1_skips_parse_without_variable_shell_flags(self) -> None: + with patch.object(tm_mod, "parse_python_source", side_effect=AssertionError("parsed")): + findings = tm_mod.analyze("subprocess.run(cmd, shell=True)", "runner.py", "python") + assert any(finding.rule_id == "TM1" for finding in findings) + + def test_tm1_preserves_nonlocal_binding_scope(self) -> None: + content = ( + "def outer():\n" + " use_shell = False\n" + " def assign():\n" + " use_shell = True\n" + " def run():\n" + " nonlocal use_shell\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_application_specific_no_verify_flag_is_not_tool_misuse(self) -> None: content = """\ print("verification: skipped (--no-verify)")