diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py index 3b457e57e..d3fdd26ba 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py +++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py @@ -17,12 +17,21 @@ from __future__ import annotations +import ast +import posixpath import re import sys from bisect import bisect_right +from contextvars import ContextVar from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Location, Severity +from skillspector.python_ast import ( + ParsedPythonFile, + parse_python_source, + peek_python_ast, + peek_python_ast_any_content, +) from skillspector.state import AnalyzerNodeResponse, SkillspectorState from . import static_runner @@ -34,6 +43,7 @@ get_context, get_context_from_lines, get_line_number, + resolve_call_name, ) from .pattern_defaults import PatternCategory @@ -41,6 +51,15 @@ ANALYZER_ID = "static_patterns_privilege_escalation" +# Scan-scoped handle on the runner's shared Python AST cache, published by +# node() for the analyze() calls the runner makes on its behalf. This module +# stays lexical (no USES_PYTHON_AST opt-in) so its windowed and normalized +# views keep running; the constructed-path analysis below consults the shared +# cache through this key instead of reparsing. +_scan_python_ast_cache_key: ContextVar[str | None] = ContextVar( + "privilege_escalation_python_ast_cache_key", default=None +) + PE1_CODE_PATTERNS = [ (r"permissions?\s*:\s*\[?\s*['\"]?\*['\"]?\s*\]?", 0.8), ( @@ -659,7 +678,179 @@ def _is_qualified_benign_access_requirement( return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements" -def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: +# Cheap pre-check before walking for constructed join calls: the AST walk +# below is only worthwhile when the text plausibly contains a join() call. +# The gate covers spelling aliases collected from the imports (``from +# os.path import join as j`` binds ``j`` to ``os.path.join``, so the call +# site reads ``j(`` with no literal ``join`` in sight). Keeping the module +# lexical plus this alias-aware gate preserves the runner's parse-once +# invariant for files without any join call. +_JOIN_CALL_BASE_NAMES = {"join"} +_JOIN_CALL_ALIAS_TARGETS = {"os.path", "os.path.join"} + +# Lexical pre-check for a windowed fragment that binds a renamed ``join`` +# without ever spelling ``join(`` (``from os.path import join as j`` calls +# ``j(``). On a whole-file cache miss the fragment is parsed to collect its +# import aliases before the alias-aware join gate below runs, so such +# fragments must also clear the parse gate; import-only fragments are +# filtered out again by the alias-aware gate. +_JOIN_IMPORT_HINT = re.compile(r"^\s*from\s+os\.path\s+import\b", re.MULTILINE) + + +def _join_call_hint(aliases: dict[str, str]) -> re.Pattern[str]: + """Return a pre-check pattern matching ``join(`` and imported join aliases. + + Every local name the file binds to ``os.path`` or ``os.path.join`` is a + possible call spelling (``j(`` for ``from os.path import join as j``), + alongside the plain ``join(`` used by ``os.path.join(``, ``p.join(``, + and direct ``join(`` imports. + """ + names = set(_JOIN_CALL_BASE_NAMES) + for local, qualified in aliases.items(): + if qualified in _JOIN_CALL_ALIAS_TARGETS: + names.add(local) + return re.compile(r"\b(?:" + "|".join(sorted(re.escape(name) for name in names)) + r")\s*\(") + + +def _fragment_whole_file_line_offset(fragment: str, whole_content: str) -> int | None: + """Return how many whole-file lines precede a windowed view fragment. + + Windowed view fragments are contiguous slices of the scanned file, so a + whole-file AST line number maps onto a fragment-relative line by + subtracting this offset. Return ``None`` when the fragment is not a + slice of the whole file (for example a normalized view), so callers keep + the standalone-parse fallback. + """ + start = whole_content.find(fragment) + if start < 0: + return None + return whole_content.count("\n", 0, start) + + +def _constructed_sensitive_paths( + content: str, + file_path: str, + python_ast: ParsedPythonFile | None = None, +) -> list[tuple[int, int, str, float]]: + """Return literal sensitive paths assembled with ``os.path.join`` in Python. + + Each hit is a ``(start_line, end_line, path, confidence)`` tuple anchored + to the call's source span so callers can deduplicate raw findings across + the whole occurrence, including calls wrapped over several lines. + Resolved from the Python AST so calls split across lines and supported + import spellings (``import os.path as p``, ``from os.path import join``, + ``from os.path import join as j``, ``from os import path``) are recognized + without reparsing tricks. The scan's shared parse is reused whenever this + runs inside the runner (the cache key published by node()); standalone + callers get a single on-demand parse. Windowed view fragments under a scan + miss the whole-file cache entry; behind the same textual join-or-import + gate they are evaluated against the whole file's cached tree instead of + being parsed standalone, so a fragment starting mid-block (for example + inside a function body, which is not a valid module on its own) keeps its + findings instead of silently dropping them. The whole file's import-alias + map is carried in from the scan cache for the join gate and call + resolution, so a renamed ``os.path.join`` spelling (``from os.path import + join as j``) is recognized even when the import lives in an earlier window + than the call. Call spans map onto fragment-relative lines and only + calls starting inside the fragment are owned by it, preserving the + runner's source-coordinate and dedupe behavior. When the whole-file tree + is unavailable, or the fragment is not a slice of the whole file + (normalized views), the fragment is parsed directly as before. Only + fully-literal positional argument lists are resolved; anything dynamic is + left to the existing pattern loop. Unparseable content simply yields no + findings here. + """ + whole_file_aliases: dict[str, str] | None = None + whole_file_tree: ast.Module | None = None + whole_file_source: str | None = None + whole_file_line_offset: int | None = None + if python_ast is None: + cache_key = _scan_python_ast_cache_key.get() + if cache_key is not None: + python_ast = peek_python_ast(cache_key, content, file_path) + if python_ast is None: + # A windowed view fragment, not the scan's whole file: the + # shared tree does not cover this slice. Carry the whole + # file's import-alias map and tree from the scan cache first + # (a fragment is a slice of the same scanned file), so the + # textual gate below also fires on renamed call spellings + # like ``j(`` when the ``from os.path import join as j`` + # import lives in an earlier window. Fragments without a + # plausible join call or join import never parse. + whole_file = peek_python_ast_any_content(cache_key, file_path) + if whole_file is not None and whole_file.tree is not None: + whole_file_aliases = whole_file.import_aliases + whole_file_tree = whole_file.tree + whole_file_source = whole_file.content + if _join_call_hint(whole_file_aliases or {}).search( + content + ) or _JOIN_IMPORT_HINT.search(content): + if whole_file_tree is not None and whole_file_source is not None: + # Evaluate the whole-file tree rather than parsing the + # fragment standalone: a slice starting mid-block is + # not a valid module, so its standalone parse fails + # and the finding would be silently dropped. + whole_file_line_offset = _fragment_whole_file_line_offset( + content, whole_file_source + ) + if whole_file_line_offset is None: + python_ast = parse_python_source(content, file_path) + else: + python_ast = parse_python_source(content, file_path) + if whole_file_tree is not None and whole_file_line_offset is not None: + tree = whole_file_tree + aliases = whole_file_aliases or {} + owned_span: tuple[int, int] | None = ( + whole_file_line_offset, + whole_file_line_offset + len(content.splitlines()), + ) + else: + if python_ast is None: + return [] + tree = python_ast.tree + if tree is None: + return [] + aliases = ( + whole_file_aliases if whole_file_aliases is not None else python_ast.import_aliases + ) + owned_span = None + if not _join_call_hint(aliases).search(content): + return [] + resolved: list[tuple[int, int, str, float]] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if resolve_call_name(node, aliases) != "os.path.join": + continue + if node.keywords or len(node.args) < 2: + continue + parts = [arg.value for arg in node.args if isinstance(arg, ast.Constant)] + if len(parts) != len(node.args) or not all(isinstance(part, str) for part in parts): + continue + value = posixpath.join(*parts) + start_line = node.lineno + end_line = node.end_lineno or node.lineno + if owned_span is not None: + # Only calls starting inside this fragment are owned by it; the + # runner's own owned-range filter and cross-window dedupe keep + # exactly one finding per call at whole-file coordinates. + if not owned_span[0] < start_line <= owned_span[1]: + continue + start_line -= owned_span[0] + end_line -= owned_span[0] + for pattern, confidence in PE3_PATTERNS: + if re.search(pattern, value, re.IGNORECASE): + resolved.append((start_line, end_line, value, confidence)) + break + return resolved + + +def analyze( + content: str, + file_path: str, + file_type: str, + python_ast: ParsedPythonFile | None = None, +) -> list[AnalyzerFinding]: """Analyze content for privilege escalation patterns (PE1–PE5).""" findings: list[AnalyzerFinding] = [] line_starts, line_ends = _source_line_metadata(content) @@ -808,6 +999,40 @@ def context_at(offset: int) -> str: complete_match=match.group(0), ) ) + if file_type == "python": + for start_line, end_line, path, confidence in _constructed_sensitive_paths( + content, file_path, python_ast + ): + constructed = AnalyzerFinding( + rule_id="PE3", + message="Credential Access", + severity=Severity.HIGH, + location=loc(start_line), + confidence=confidence, + tags=list(tag), + context=get_context(content, line_starts[start_line - 1]), + matched_text=path, + ) + # One PE3 per source occurrence: the pattern loop above may already + # have fired inside the join call's line span (for example the + # literal '.ssh/id_rsa' on a wrapped argument line, while this + # finding anchors to the call's first line). Keep the + # best-confidence finding per call span, mirroring the PE4/PE5 + # per-line aggregation below; unrelated occurrences on other + # lines are preserved. + duplicate = next( + ( + existing + for existing in findings + if existing.rule_id == "PE3" + and start_line <= existing.location.start_line <= end_line + ), + None, + ) + if duplicate is None: + findings.append(constructed) + elif confidence > duplicate.confidence: + findings[findings.index(duplicate)] = constructed # Collect best-confidence PE4 finding per line to avoid double-counting lines # that match multiple patterns (e.g. DockerClient(base_url=".../docker.sock")). pe4_best: dict[int, AnalyzerFinding] = {} @@ -955,6 +1180,10 @@ def _is_negated_safety_constraint( def node(state: SkillspectorState) -> AnalyzerNodeResponse: """Run privilege_escalation patterns and return findings.""" - response = static_runner.run_static_patterns_with_ledger(state, [sys.modules[__name__]]) + token = _scan_python_ast_cache_key.set(state.get("python_ast_cache_key")) + try: + response = static_runner.run_static_patterns_with_ledger(state, [sys.modules[__name__]]) + finally: + _scan_python_ast_cache_key.reset(token) logger.info("%s: %d findings", ANALYZER_ID, len(response["findings"])) return response diff --git a/src/skillspector/python_ast.py b/src/skillspector/python_ast.py index 2bf1ce5ca..2637615a2 100644 --- a/src/skillspector/python_ast.py +++ b/src/skillspector/python_ast.py @@ -337,6 +337,46 @@ def get_python_ast(cache_key: str | None, content: str, filename: str) -> Parsed return parsed +def peek_python_ast(cache_key: str | None, content: str, filename: str) -> ParsedPythonFile | None: + """Return the scan's cached parse for *filename* without parsing anything. + + Only an exact content match hits. Anything else (a windowed view fragment, + an uncached file, an unknown cache key) returns ``None`` so the caller can + apply its own fallback without disturbing the shared cache: unlike + :func:`get_python_ast`, this never parses and never stores. + """ + if cache_key is None: + return None + with _runtime_ast_cache_lock: + cache = _runtime_ast_caches.get(cache_key) + if cache is None: + return None + cached = cache.entries.get(filename) + if cached is not None and cached.content == content: + cache.entries.move_to_end(filename) + return cached + return None + + +def peek_python_ast_any_content(cache_key: str | None, filename: str) -> ParsedPythonFile | None: + """Return the scan's cached parse for *filename* regardless of content. + + Windowed view fragments never content-match the whole-file entry, but a + fragment is a slice of the same scanned file, so its import-alias map is + still valid for resolving calls that the fragment's own imports cannot + explain (for example a renamed ``os.path.join`` import living in an + earlier window). Like :func:`peek_python_ast`, this never parses and + never stores. + """ + if cache_key is None: + return None + with _runtime_ast_cache_lock: + cache = _runtime_ast_caches.get(cache_key) + if cache is None: + return None + return cache.entries.get(filename) + + def clear_python_ast_cache(cache_key: str | None) -> None: """Release one scan's process-local parsed trees after its analyzer phase.""" if cache_key is None: diff --git a/tests/nodes/analyzers/test_shared_python_ast.py b/tests/nodes/analyzers/test_shared_python_ast.py index fa864ed60..052a52cfe 100644 --- a/tests/nodes/analyzers/test_shared_python_ast.py +++ b/tests/nodes/analyzers/test_shared_python_ast.py @@ -16,6 +16,7 @@ behavioral_taint_tracking, static_patterns_data_exfiltration, static_patterns_output_handling, + static_runner, ) from skillspector.nodes.build_context import build_context from skillspector.nodes.deduplicate import deduplicate @@ -128,12 +129,18 @@ def count_parse(*args, **kwargs): def test_graph_scan_parses_python_once_before_parallel_analyzers(tmp_path, monkeypatch) -> None: - """The runtime cache shares one parse across the graph's analyzer fan-out.""" + """The runtime cache shares one parse across the graph's analyzer fan-out. + + The fixture includes a literal ``os.path.join`` call so the test proves + the supplemental constructed-path analysis reuses the shared parse rather + than parsing again. + """ (tmp_path / "script.py").write_text( "import os\n" "import subprocess\n" "payload = input()\n" "environment = os.environ.copy()\n" + "credential = os.path.join('/etc', 'passwd')\n" "subprocess.run(output)\n" "exec(payload)\n", encoding="utf-8", @@ -151,5 +158,36 @@ def count_parse(*args, **kwargs): result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) assert {"E2", "OH1", "AST1", "TT5"} <= {finding.rule_id for finding in result["findings"]} + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in result["findings"] + ) assert parse_calls == 1 assert JsonPlusSerializer().dumps_typed(result) + + +def test_graph_scan_reports_constructed_path_above_view_window_chars(tmp_path) -> None: + """Windowed lexical scans keep constructed-path PE3 above the view window. + + Regression test: routing the constructed-path analysis through + ``peek_python_ast`` dropped findings once the runner sliced content into + window views (above ``SECURITY_VIEW_WINDOW_CHARS``), because a slice never + matches the scan's whole-file cache entry. The fragment fallback parses + the slice directly so large files keep their findings. The source spells + the call through a renamed import (``from os.path import join as j``) to + pin that the cache-miss fallback recognizes renamed join spellings: the + plain ``join(`` textual gate never fires on the ``j(`` call site. + """ + filler_line = "# " + "x" * 118 + "\n" + body = "from os.path import join as j\ncredential = j('/etc', 'passwd')\n" + target_chars = static_runner.SECURITY_VIEW_WINDOW_CHARS + 120_000 + source = body + filler_line * ((target_chars - len(body)) // len(filler_line)) + assert len(source) > static_runner.SECURITY_VIEW_WINDOW_CHARS + (tmp_path / "large_script.py").write_text(source, encoding="utf-8") + + result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in result["findings"] + ) diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py index e7d03b74e..6080d1eee 100644 --- a/tests/unit/test_patterns.py +++ b/tests/unit/test_patterns.py @@ -19,6 +19,7 @@ import pytest +import skillspector.python_ast as python_ast_module from skillspector.models import Severity from skillspector.nodes.analyzers import ( static_patterns_data_exfiltration as data_exfiltration_module, @@ -36,6 +37,7 @@ static_patterns_supply_chain as supply_chain_module, ) from skillspector.nodes.analyzers import static_runner +from skillspector.python_ast import prewarm_python_ast_cache def _assert_contextual_pe3(findings) -> None: @@ -287,6 +289,308 @@ def test_pe3_env_file(self) -> None: findings = privilege_escalation_module.analyze(content, "SKILL.md", "markdown") assert len(findings) >= 1 + def test_pe3_constant_os_path_join_passwd(self) -> None: + """A statically constructed credential path must retain PE3 coverage.""" + content = ( + "import os\n" + "path = os.path.join('/etc', 'passwd')\n" + "with open(path) as source:\n" + " data = source.read()\n" + ) + + findings = privilege_escalation_module.analyze(content, "exploit.py", "python") + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ) + + def test_pe3_constructed_join_is_deduplicated_per_line(self) -> None: + """One source occurrence must not produce two PE3 findings for a line.""" + content = ( + "import os\n" + "path = os.path.join('.ssh/id_rsa', 'x')\n" + "with open(path) as source:\n" + " data = source.read()\n" + ) + + findings = privilege_escalation_module.analyze(content, "exploit.py", "python") + + line_two = [f for f in findings if f.rule_id == "PE3" and f.location.start_line == 2] + assert len(line_two) == 1 + + def test_pe3_multiline_constructed_join_is_deduplicated(self) -> None: + """A join call wrapped across lines must yield one PE3 for its span.""" + content = ( + "import os\n" + "p = os.path.join(\n" + " '.ssh/id_rsa', 'x'\n" + ")\n" + "with open(p) as source:\n" + " data = source.read()\n" + ) + + findings = privilege_escalation_module.analyze(content, "exploit.py", "python") + + span_pe3 = [f for f in findings if f.rule_id == "PE3" and 2 <= f.location.start_line <= 4] + assert len(span_pe3) == 1 + assert span_pe3[0].message == "Credential Access" + assert span_pe3[0].severity == Severity.HIGH + + def test_pe3_multiline_os_path_join_is_detected(self) -> None: + """A join call wrapped across lines must retain PE3 coverage.""" + content = ( + "import os\n" + "path = os.path.join(\n" + " '/etc', 'passwd'\n" + ")\n" + "with open(path) as source:\n" + " data = source.read()\n" + ) + + findings = privilege_escalation_module.analyze(content, "exploit.py", "python") + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ) + + def test_pe3_aliased_join_imports_are_detected(self) -> None: + """Supported import spellings of os.path.join must retain PE3 coverage.""" + for header in ( + "from os.path import join\n", + "from os.path import join as j\n", + "import os.path as p\n", + "from os import path\n", + ): + call = { + "from os.path import join\n": "join('/etc', 'passwd')\n", + "from os.path import join as j\n": "j('/etc', 'passwd')\n", + "import os.path as p\n": "p.join('/etc', 'passwd')\n", + "from os import path\n": "path.join('/etc', 'passwd')\n", + }[header] + content = header + "target = " + call + + findings = privilege_escalation_module.analyze(content, "exploit.py", "python") + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ), header + + def test_pe3_windowed_fragment_keeps_constructed_path_coverage(self) -> None: + """A windowed view fragment under a scan must retain PE3 coverage. + + Regression test: above SECURITY_VIEW_WINDOW_CHARS the runner hands + lexical modules window slices, which never match the scan's + whole-file AST cache entry. The constructed-path analysis must parse + the fragment directly instead of silently dropping its findings. + The fragment omits the import line, as a later window would, to pin + that the dotted ``os.path.join`` spelling resolves without aliases. + """ + whole = "import os\npath = os.path.join('/etc', 'passwd')\nprint(path)\n" + # A later window slice as the runner would hand it: complete and + # parseable, but missing the import line and unable to match the + # whole-file cache entry. + fragment = "".join(whole.splitlines(keepends=True)[1:]) + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": whole}) + token = privilege_escalation_module._scan_python_ast_cache_key.set(cache_key) + try: + findings = privilege_escalation_module.analyze(fragment, "exploit.py", "python") + finally: + privilege_escalation_module._scan_python_ast_cache_key.reset(token) + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ) + + def test_pe3_windowed_fragment_with_renamed_join_import_keeps_coverage(self) -> None: + """A windowed fragment calling a renamed join import retains PE3 coverage. + + Regression test: the cache-miss fallback gated on the plain ``join(`` + spelling, so a fragment spelling the call ``j(`` (bound by ``from + os.path import join as j``) silently dropped its finding. The + fallback now also parses fragments importing ``os.path.join`` so the + alias-aware gate sees the renamed spelling. The fragment keeps the + import line: a renamed call site never spells ``join(``, so only the + import can clear the fragment parse gate. + """ + whole = ( + "from os.path import join as j\ncredential = j('/etc', 'passwd')\nprint(credential)\n" + ) + # A later window slice as the runner would hand it: complete and + # parseable, but unable to match the whole-file cache entry. The + # import line is retained here because a renamed call site never + # spells ``join(``, so only the import can clear the parse gate. + fragment = "".join(whole.splitlines(keepends=True)[:2]) + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": whole}) + token = privilege_escalation_module._scan_python_ast_cache_key.set(cache_key) + try: + findings = privilege_escalation_module.analyze(fragment, "exploit.py", "python") + finally: + privilege_escalation_module._scan_python_ast_cache_key.reset(token) + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ), fragment + + def test_pe3_windowed_fragment_with_import_in_other_window_keeps_coverage(self) -> None: + """A fragment whose join import lives in another window retains PE3 coverage. + + Regression test: the import and the call fall in different raw + windows, so the call's fragment carries no import aliases of its own + and never spells ``join(``. The constructed-path analysis must carry + the whole file's import-alias map from the scan cache for the join + gate and call resolution instead of silently dropping the finding. + """ + whole = ( + "from os.path import join as j\ncredential = j('/etc', 'passwd')\nprint(credential)\n" + ) + # The later window slice as the runner would hand it: complete and + # parseable, but the import line lives in the earlier window and the + # slice cannot match the whole-file cache entry. + fragment = "".join(whole.splitlines(keepends=True)[1:]) + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": whole}) + token = privilege_escalation_module._scan_python_ast_cache_key.set(cache_key) + try: + findings = privilege_escalation_module.analyze(fragment, "exploit.py", "python") + finally: + privilege_escalation_module._scan_python_ast_cache_key.reset(token) + + assert any( + finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + for finding in findings + ), fragment + + def test_pe3_graph_windowed_import_and_call_across_windows(self) -> None: + """The full node path keeps PE3 coverage when import and call split windows. + + Graph regression for the reviewer finding on the current head: the + source is large enough that the runner splits it into two raw + windows, the first holding the ``from os.path import join as j`` + import but no call, and the second holding ``j('/etc', 'passwd')`` + but neither the import nor a literal ``join(``. The constructed + sensitive path must still be reported exactly once. + """ + padding_line = "# " + "x" * 118 + "\n" + pad_lines = static_runner.SECURITY_VIEW_WINDOW_CHARS // len(padding_line) + 10 + content = ( + "from os.path import join as j\n" + + padding_line * pad_lines + + "credential = j('/etc', 'passwd')\n" + ) + assert len(content) > static_runner.SECURITY_VIEW_WINDOW_CHARS + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": content}) + response = privilege_escalation_module.node( + { + "components": ["exploit.py"], + "file_cache": {"exploit.py": content}, + "python_ast_cache_key": cache_key, + } + ) + constructed = [ + finding + for finding in response["findings"] + if finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + ] + assert len(constructed) == 1 + + def test_pe3_windowed_fragment_inside_function_body_keeps_coverage(self) -> None: + """A fragment starting inside a function body keeps PE3 coverage. + + Regression test: the fallback parsed each windowed fragment as a + standalone module, so a slice starting mid-block (indented, with the + enclosing ``def`` in an earlier window) failed parsing and silently + dropped its findings. The fallback now evaluates the whole-file + tree with spans mapped onto fragment lines, so the constructed + sensitive path is reported exactly once at its fragment location. + """ + padding = (" # " + "x" * 118 + "\n") * 40 + whole = ( + "from os.path import join as j\n" + "def load():\n" + " pass\n" + padding + " credential = j('/etc', 'passwd')\n" + ) + lines = whole.splitlines(keepends=True) + # A later window slice as the runner would hand it: starts inside the + # function body, so it cannot parse as a standalone module, and the + # import plus the ``def`` line live in the earlier window. + fragment = "".join(lines[10:]) + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": whole}) + token = privilege_escalation_module._scan_python_ast_cache_key.set(cache_key) + try: + findings = privilege_escalation_module.analyze(fragment, "exploit.py", "python") + finally: + privilege_escalation_module._scan_python_ast_cache_key.reset(token) + + constructed = [ + finding + for finding in findings + if finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + ] + assert len(constructed) == 1 + assert constructed[0].location.start_line == len(lines) - 10 + + def test_pe3_graph_windowed_call_inside_function_keeps_coverage(self) -> None: + """The full node path keeps PE3 coverage for a call inside a function. + + Graph regression for the reviewer finding on the current head: the + source is large enough that the runner splits it into two raw + windows, and the second window starts inside the ``load`` function + body, so parsing that slice as a standalone module fails. The + constructed sensitive path must still be reported exactly once, at + the call's original whole-file location. + """ + padding_line = " # " + "x" * 118 + "\n" + pad_lines = static_runner.SECURITY_VIEW_WINDOW_CHARS // len(padding_line) + 10 + content = ( + "from os.path import join as j\n" + "def load():\n" + " pass\n" + padding_line * pad_lines + " credential = j('/etc', 'passwd')\n" + ) + assert len(content) > static_runner.SECURITY_VIEW_WINDOW_CHARS + call_line = 3 + pad_lines + 1 + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": content}) + response = privilege_escalation_module.node( + { + "components": ["exploit.py"], + "file_cache": {"exploit.py": content}, + "python_ast_cache_key": cache_key, + } + ) + constructed = [ + finding + for finding in response["findings"] + if finding.rule_id == "PE3" and finding.matched_text == "/etc/passwd" + ] + assert len(constructed) == 1 + assert constructed[0].start_line == call_line + + def test_pe3_windowed_fragment_without_join_call_does_not_parse(self, monkeypatch) -> None: + """Fragments without a plausible join call must not pay for a parse.""" + whole = "import os\npath = os.path.join('/etc', 'passwd')\n" + fragment = "# just a comment line\nx = 1\n" + cache_key = prewarm_python_ast_cache(["exploit.py"], {"exploit.py": whole}) + + parse_calls = 0 + original_parse = python_ast_module.ast.parse + + def count_parse(*args, **kwargs): + nonlocal parse_calls + parse_calls += 1 + return original_parse(*args, **kwargs) + + monkeypatch.setattr(python_ast_module.ast, "parse", count_parse) + token = privilege_escalation_module._scan_python_ast_cache_key.set(cache_key) + try: + privilege_escalation_module.analyze(fragment, "exploit.py", "python") + finally: + privilege_escalation_module._scan_python_ast_cache_key.reset(token) + + assert parse_calls == 0 + # -- PE3 false-positive prevention -- def test_pe3_gitlab_settings_access_tokens_is_contextualized(self) -> None: