From b84ea49757368e10dd67982ec8cbc27262b9535e Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Mon, 14 Sep 2026 13:15:37 -0700
Subject: [PATCH 1/9] fix(patterns): detect constructed sensitive paths

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 54 +++++++++++++++++++
 tests/unit/test_patterns.py                   | 16 ++++++
 2 files changed, 70 insertions(+)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 3b457e57e..4b7e7fc3d 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+import ast
+import posixpath
 import re
 import sys
 from bisect import bisect_right
@@ -659,6 +661,44 @@ def _is_qualified_benign_access_requirement(
     return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements"
 
 
+def _constant_os_path_join(node: ast.expr) -> str | None:
+    """Resolve literal ``os.path.join`` calls without evaluating arbitrary code."""
+    if not (
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "join"
+        and isinstance(node.func.value, ast.Attribute)
+        and node.func.value.attr == "path"
+        and isinstance(node.func.value.value, ast.Name)
+        and node.func.value.value.id == "os"
+    ):
+        return None
+
+    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):
+        return None
+    return posixpath.join(*parts)
+
+
+def _constructed_sensitive_paths(content: str) -> list[tuple[int, str, float]]:
+    """Return literal sensitive paths assembled with ``os.path.join`` in Python."""
+    try:
+        tree = ast.parse(content)
+    except SyntaxError:
+        return []
+
+    resolved: list[tuple[int, str, float]] = []
+    for node in ast.walk(tree):
+        value = _constant_os_path_join(node)
+        if value is None:
+            continue
+        for pattern, confidence in PE3_PATTERNS:
+            if re.search(pattern, value, re.IGNORECASE):
+                resolved.append((node.lineno, value, confidence))
+                break
+    return resolved
+
+
 def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]:
     """Analyze content for privilege escalation patterns (PE1–PE5)."""
     findings: list[AnalyzerFinding] = []
@@ -808,6 +848,20 @@ def context_at(offset: int) -> str:
                     complete_match=match.group(0),
                 )
             )
+    if file_type == "python":
+        for line_num, path, confidence in _constructed_sensitive_paths(content):
+            findings.append(
+                AnalyzerFinding(
+                    rule_id="PE3",
+                    message="Credential Access",
+                    severity=Severity.HIGH,
+                    location=loc(line_num),
+                    confidence=confidence,
+                    tags=list(tag),
+                    context=get_context(content, line_starts[line_num - 1]),
+                    matched_text=path,
+                )
+            )
     # 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] = {}
diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py
index e7d03b74e..593108a15 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -287,6 +287,22 @@ 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
+        )
+
     # -- PE3 false-positive prevention --
 
     def test_pe3_gitlab_settings_access_tokens_is_contextualized(self) -> None:

From 17822796541a7eac2d03d2d898a72b546d8768a1 Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Mon, 14 Sep 2026 15:14:26 -0700
Subject: [PATCH 2/9] fix(patterns): preserve shared Python AST cache

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 43 ++++++-------------
 1 file changed, 14 insertions(+), 29 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 4b7e7fc3d..6d597d73d 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -17,7 +17,6 @@
 
 from __future__ import annotations
 
-import ast
 import posixpath
 import re
 import sys
@@ -661,40 +660,26 @@ def _is_qualified_benign_access_requirement(
     return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements"
 
 
-def _constant_os_path_join(node: ast.expr) -> str | None:
-    """Resolve literal ``os.path.join`` calls without evaluating arbitrary code."""
-    if not (
-        isinstance(node, ast.Call)
-        and isinstance(node.func, ast.Attribute)
-        and node.func.attr == "join"
-        and isinstance(node.func.value, ast.Attribute)
-        and node.func.value.attr == "path"
-        and isinstance(node.func.value.value, ast.Name)
-        and node.func.value.value.id == "os"
-    ):
-        return None
-
-    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):
-        return None
-    return posixpath.join(*parts)
-
-
 def _constructed_sensitive_paths(content: str) -> list[tuple[int, str, float]]:
-    """Return literal sensitive paths assembled with ``os.path.join`` in Python."""
-    try:
-        tree = ast.parse(content)
-    except SyntaxError:
-        return []
+    """Return literal sensitive paths assembled with ``os.path.join`` in Python.
+
+    Keep this expression-level recognizer regex-based: Python AST parsing is shared
+    across the analyzer graph, so a second parse here would defeat that cache.
+    """
+    call_pattern = re.compile(r"os\.path\.join\((?P<args>[^()\n]+)\)")
+    string_pattern = re.compile(r"(['\"])(?P<value>[^'\"]*)\1")
 
     resolved: list[tuple[int, str, float]] = []
-    for node in ast.walk(tree):
-        value = _constant_os_path_join(node)
-        if value is None:
+    for match in call_pattern.finditer(content):
+        args = match.group("args")
+        parts = [item.group("value") for item in string_pattern.finditer(args)]
+        residual = string_pattern.sub("", args).replace(",", "").strip()
+        if len(parts) < 2 or residual:
             continue
+        value = posixpath.join(*parts)
         for pattern, confidence in PE3_PATTERNS:
             if re.search(pattern, value, re.IGNORECASE):
-                resolved.append((node.lineno, value, confidence))
+                resolved.append((get_line_number(content, match.start()), value, confidence))
                 break
     return resolved
 

From 826273706a597295a3ce9c2e787478d91621e9aa Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Sun, 20 Sep 2026 05:33:51 +0000
Subject: [PATCH 3/9] fix(patterns): resolve constructed joins via AST with
 per-line PE3 dedup

The PE3 constructed-path recognizer missed multiline and imported/aliased
os.path.join spellings, and emitted a second PE3 for lines the pattern loop
already covered. Resolve literal join calls from the Python AST (handling
from/import aliases), and keep the best-confidence PE3 per line.

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 76 +++++++++++++------
 tests/unit/test_patterns.py                   | 53 +++++++++++++
 2 files changed, 105 insertions(+), 24 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 6d597d73d..ddf21833e 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import ast
 import posixpath
 import re
 import sys
@@ -24,6 +25,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.state import AnalyzerNodeResponse, SkillspectorState
 
 from . import static_runner
@@ -35,6 +37,7 @@
     get_context,
     get_context_from_lines,
     get_line_number,
+    resolve_call_name,
 )
 from .pattern_defaults import PatternCategory
 
@@ -660,26 +663,37 @@ def _is_qualified_benign_access_requirement(
     return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements"
 
 
-def _constructed_sensitive_paths(content: str) -> list[tuple[int, str, float]]:
+def _constructed_sensitive_paths(content: str, file_path: str) -> list[tuple[int, str, float]]:
     """Return literal sensitive paths assembled with ``os.path.join`` in Python.
 
-    Keep this expression-level recognizer regex-based: Python AST parsing is shared
-    across the analyzer graph, so a second parse here would defeat that cache.
+    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 import path``) are recognized without reparsing tricks.  Only
+    fully-literal positional argument lists are resolved; anything dynamic is
+    left to the existing pattern loop.  Parsing stays local to this module so
+    its runner contract (lexical scanning, windowed views) does not change;
+    unparseable content simply yields no findings here.
     """
-    call_pattern = re.compile(r"os\.path\.join\((?P<args>[^()\n]+)\)")
-    string_pattern = re.compile(r"(['\"])(?P<value>[^'\"]*)\1")
-
+    parsed = parse_python_source(content, file_path)
+    tree = parsed.tree
+    if tree is None:
+        return []
+    aliases = parsed.import_aliases
     resolved: list[tuple[int, str, float]] = []
-    for match in call_pattern.finditer(content):
-        args = match.group("args")
-        parts = [item.group("value") for item in string_pattern.finditer(args)]
-        residual = string_pattern.sub("", args).replace(",", "").strip()
-        if len(parts) < 2 or residual:
+    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)
         for pattern, confidence in PE3_PATTERNS:
             if re.search(pattern, value, re.IGNORECASE):
-                resolved.append((get_line_number(content, match.start()), value, confidence))
+                resolved.append((node.lineno, value, confidence))
                 break
     return resolved
 
@@ -834,19 +848,33 @@ def context_at(offset: int) -> str:
                 )
             )
     if file_type == "python":
-        for line_num, path, confidence in _constructed_sensitive_paths(content):
-            findings.append(
-                AnalyzerFinding(
-                    rule_id="PE3",
-                    message="Credential Access",
-                    severity=Severity.HIGH,
-                    location=loc(line_num),
-                    confidence=confidence,
-                    tags=list(tag),
-                    context=get_context(content, line_starts[line_num - 1]),
-                    matched_text=path,
-                )
+        for line_num, path, confidence in _constructed_sensitive_paths(content, file_path):
+            constructed = AnalyzerFinding(
+                rule_id="PE3",
+                message="Credential Access",
+                severity=Severity.HIGH,
+                location=loc(line_num),
+                confidence=confidence,
+                tags=list(tag),
+                context=get_context(content, line_starts[line_num - 1]),
+                matched_text=path,
+            )
+            # One PE3 per source occurrence: the pattern loop above may already
+            # have fired on this line's raw text (for example the literal
+            # '.ssh/id_rsa' inside the join call).  Keep the best-confidence
+            # finding, mirroring the PE4/PE5 per-line aggregation below.
+            duplicate = next(
+                (
+                    existing
+                    for existing in findings
+                    if existing.rule_id == "PE3" and existing.location.start_line == line_num
+                ),
+                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] = {}
diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py
index 593108a15..65b7cd794 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -303,6 +303,59 @@ def test_pe3_constant_os_path_join_passwd(self) -> None:
             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_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",
+            "import os.path as p\n",
+            "from os import path\n",
+        ):
+            call = {
+                "from os.path import join\n": "join('/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
+
     # -- PE3 false-positive prevention --
 
     def test_pe3_gitlab_settings_access_tokens_is_contextualized(self) -> None:

From 40cea883f8019ff0db6a15edd5b7f916aac1ab4f Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Sun, 20 Sep 2026 05:59:36 +0000
Subject: [PATCH 4/9] fix(patterns): gate constructed-path AST parse behind
 join pre-check

The AST-based os.path.join resolver parsed every Python file, defeating
the runner's parse-once invariant (one shared parse per file across the
analyzer fan-out). Gate the local parse behind a cheap textual pre-check
so files without a plausible join() call never pay for it. The module
stays lexical, preserving windowed and normalized-view scans for all
file types.

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py       | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index ddf21833e..82cf576cb 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -663,6 +663,14 @@ def _is_qualified_benign_access_requirement(
     return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements"
 
 
+# Cheap pre-check before parsing for constructed join calls: the AST walk
+# below is only worthwhile when the text plausibly contains a join() call.
+# This keeps the module lexical (windowed and normalized-view scans keep
+# working for every file type) while preserving the runner's parse-once
+# invariant for files without any join call.
+_JOIN_CALL_HINT = re.compile(r"\bjoin\s*\(")
+
+
 def _constructed_sensitive_paths(content: str, file_path: str) -> list[tuple[int, str, float]]:
     """Return literal sensitive paths assembled with ``os.path.join`` in Python.
 
@@ -670,10 +678,11 @@ def _constructed_sensitive_paths(content: str, file_path: str) -> list[tuple[int
     import spellings (``import os.path as p``, ``from os.path import join``,
     ``from os import path``) are recognized without reparsing tricks.  Only
     fully-literal positional argument lists are resolved; anything dynamic is
-    left to the existing pattern loop.  Parsing stays local to this module so
-    its runner contract (lexical scanning, windowed views) does not change;
-    unparseable content simply yields no findings here.
+    left to the existing pattern loop.  Unparseable content simply yields no
+    findings here.
     """
+    if not _JOIN_CALL_HINT.search(content):
+        return []
     parsed = parse_python_source(content, file_path)
     tree = parsed.tree
     if tree is None:

From 56d121cae8595f8203ab3b52be34f953c1f698bd Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Tue, 22 Sep 2026 01:11:28 +0000
Subject: [PATCH 5/9] fix(patterns): alias-aware join gate, shared AST reuse,
 span dedup

Address rng1995's review on constructed sensitive-path detection:

- Broaden the join pre-check to cover imported local names: the gate now
  matches every name the file binds to os.path or os.path.join, so
  'from os.path import join as j' followed by j('/etc', 'passwd') reaches
  alias resolution instead of returning early on the missing literal join(.
- Route the supplemental analysis through the scan's shared AST: node()
  publishes the runner's python_ast_cache_key via a contextvar and
  _constructed_sensitive_paths consults it with the new peek_python_ast
  helper (exact-content hit only, never parses or writes). The module stays
  lexical so windowed and normalized views keep running; windowed view
  fragments are skipped and standalone callers keep one on-demand parse.
- Deduplicate across the call's whole source span: hits now carry
  (start_line, end_line), so a join call wrapped over several lines no
  longer produces a second PE3 when the lexical loop fires on a wrapped
  argument line.

Tests: extend test_pe3_aliased_join_imports_are_detected with the renamed
import case, add test_pe3_multiline_constructed_join_is_deduplicated, and
extend test_graph_scan_parses_python_once_before_parallel_analyzers with a
literal os.path.join call to verify the graph still performs one parse.

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 116 ++++++++++++++----
 src/skillspector/python_ast.py                |  21 ++++
 .../nodes/analyzers/test_shared_python_ast.py |  12 +-
 tests/unit/test_patterns.py                   |  20 +++
 4 files changed, 142 insertions(+), 27 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 82cf576cb..c46910f2e 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -22,10 +22,11 @@
 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 parse_python_source
+from skillspector.python_ast import ParsedPythonFile, parse_python_source, peek_python_ast
 from skillspector.state import AnalyzerNodeResponse, SkillspectorState
 
 from . import static_runner
@@ -45,6 +46,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),
     (
@@ -663,32 +673,71 @@ def _is_qualified_benign_access_requirement(
     return heading_index >= 0 and lines[heading_index].strip() == "## Access Requirements"
 
 
-# Cheap pre-check before parsing for constructed join calls: the AST walk
+# Cheap pre-check before walking for constructed join calls: the AST walk
 # below is only worthwhile when the text plausibly contains a join() call.
-# This keeps the module lexical (windowed and normalized-view scans keep
-# working for every file type) while preserving the runner's parse-once
+# 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_HINT = re.compile(r"\bjoin\s*\(")
+_JOIN_CALL_BASE_NAMES = {"join"}
+_JOIN_CALL_ALIAS_TARGETS = {"os.path", "os.path.join"}
+
+
+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 _constructed_sensitive_paths(content: str, file_path: str) -> list[tuple[int, str, float]]:
+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 import path``) are recognized without reparsing tricks.  Only
-    fully-literal positional argument lists are resolved; anything dynamic is
-    left to the existing pattern loop.  Unparseable content simply yields no
-    findings here.
+    ``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, while windowed view fragments under
+    a scan are skipped because the shared tree only covers whole files.
+    Only fully-literal positional argument lists are resolved; anything
+    dynamic is left to the existing pattern loop.  Unparseable content simply
+    yields no findings here.
     """
-    if not _JOIN_CALL_HINT.search(content):
-        return []
-    parsed = parse_python_source(content, file_path)
-    tree = parsed.tree
+    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 it, and parsing fragments here
+                # would break the runner's parse-once invariant.
+                return []
+        else:
+            python_ast = parse_python_source(content, file_path)
+    tree = python_ast.tree
     if tree is None:
         return []
-    aliases = parsed.import_aliases
-    resolved: list[tuple[int, str, float]] = []
+    aliases = python_ast.import_aliases
+    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
@@ -702,12 +751,17 @@ def _constructed_sensitive_paths(content: str, file_path: str) -> list[tuple[int
         value = posixpath.join(*parts)
         for pattern, confidence in PE3_PATTERNS:
             if re.search(pattern, value, re.IGNORECASE):
-                resolved.append((node.lineno, value, confidence))
+                resolved.append((node.lineno, node.end_lineno or node.lineno, value, confidence))
                 break
     return resolved
 
 
-def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]:
+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)
@@ -857,26 +911,32 @@ def context_at(offset: int) -> str:
                 )
             )
     if file_type == "python":
-        for line_num, path, confidence in _constructed_sensitive_paths(content, file_path):
+        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(line_num),
+                location=loc(start_line),
                 confidence=confidence,
                 tags=list(tag),
-                context=get_context(content, line_starts[line_num - 1]),
+                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 on this line's raw text (for example the literal
-            # '.ssh/id_rsa' inside the join call).  Keep the best-confidence
-            # finding, mirroring the PE4/PE5 per-line aggregation below.
+            # 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 existing.location.start_line == line_num
+                    if existing.rule_id == "PE3"
+                    and start_line <= existing.location.start_line <= end_line
                 ),
                 None,
             )
@@ -1031,6 +1091,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..a26a3c4c0 100644
--- a/src/skillspector/python_ast.py
+++ b/src/skillspector/python_ast.py
@@ -337,6 +337,27 @@ 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 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..c3ac08c21 100644
--- a/tests/nodes/analyzers/test_shared_python_ast.py
+++ b/tests/nodes/analyzers/test_shared_python_ast.py
@@ -128,12 +128,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 +157,9 @@ 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)
diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py
index 65b7cd794..ab2ce6164 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -317,6 +317,24 @@ def test_pe3_constructed_join_is_deduplicated_per_line(self) -> None:
         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 = (
@@ -339,11 +357,13 @@ 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]

From c8d201b978112dd0c8409c1f6ad47e8e2a616d27 Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Tue, 22 Sep 2026 03:03:52 +0000
Subject: [PATCH 6/9] fix(patterns): restore constructed-path findings on
 windowed views

Above SECURITY_VIEW_WINDOW_CHARS the runner hands lexical modules window
slices instead of the whole file, so peek_python_ast never hits the scan's
whole-file cache entry and the constructed-path analysis silently dropped
its findings (regression introduced in 56d121c; 40cea88 reported them).

On a peek miss, parse the provided content directly behind the same plain
textual join gate, restoring the 40cea88 behavior for window slices while
keeping the 17:45 improvements for whole files: alias-aware gate, shared
AST reuse, multiline span dedup, and scan-count discipline (fragments
without a plausible join call never parse).

Regression tests:
- unit: a windowed fragment under a scan keeps PE3 coverage, and a
  join-free fragment triggers no parse
- graph: a ~376k-char file keeps its constructed-path PE3 through the
  full scan

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 24 +++++----
 .../nodes/analyzers/test_shared_python_ast.py | 25 +++++++++
 tests/unit/test_patterns.py                   | 52 +++++++++++++++++++
 3 files changed, 92 insertions(+), 9 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index c46910f2e..98cd32712 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -714,23 +714,29 @@ def _constructed_sensitive_paths(
     ``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, while windowed view fragments under
-    a scan are skipped because the shared tree only covers whole files.
-    Only fully-literal positional argument lists are resolved; anything
-    dynamic is left to the existing pattern loop.  Unparseable content simply
-    yields no findings here.
+    callers get a single on-demand parse.  Windowed view fragments under a scan
+    miss the whole-file cache entry and are parsed directly behind the same
+    textual join gate, so large files keep their findings instead of silently
+    dropping them.  Only fully-literal positional argument lists are resolved;
+    anything dynamic is left to the existing pattern loop.  Unparseable content
+    simply yields no findings here.
     """
     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:
+            if python_ast is None and _join_call_hint({}).search(content):
                 # A windowed view fragment, not the scan's whole file: the
-                # shared tree does not cover it, and parsing fragments here
-                # would break the runner's parse-once invariant.
-                return []
+                # shared tree does not cover this slice, so parse the
+                # fragment directly to avoid silently dropping large-file
+                # findings.  The plain textual gate keeps the runner's
+                # scan-count discipline for fragments without a plausible
+                # join call.
+                python_ast = parse_python_source(content, file_path)
         else:
             python_ast = parse_python_source(content, file_path)
+    if python_ast is None:
+        return []
     tree = python_ast.tree
     if tree is None:
         return []
diff --git a/tests/nodes/analyzers/test_shared_python_ast.py b/tests/nodes/analyzers/test_shared_python_ast.py
index c3ac08c21..7ff47d8b0 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
@@ -163,3 +164,27 @@ def count_parse(*args, **kwargs):
     )
     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.
+    """
+    filler_line = "# " + "x" * 118 + "\n"
+    body = "import os\ncredential = os.path.join('/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 ab2ce6164..612c0c1e0 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:
@@ -376,6 +378,56 @@ def test_pe3_aliased_join_imports_are_detected(self) -> None:
                 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_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:

From cb4f9e15f851823a4b6d4033590f27f255905244 Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Tue, 22 Sep 2026 17:02:24 +0000
Subject: [PATCH 7/9] fix(patterns): recognize renamed join imports on windowed
 cache misses

The windowed-fragment fallback gated on the plain join( spelling, so a
fragment calling j( bound by from os.path import join as j never parsed
and silently dropped its constructed-path PE3 (review P1).

The parse gate now also fires on from os.path import spellings: the
fragment is parsed to collect its import aliases, and the existing
alias-aware join gate filters import-only fragments.  Fragments without
a plausible join call or join import still never parse.

Regression tests:
- graph: the ~376k-char large-file regression now spells the call
  through the renamed import with trailing comment padding
- unit: a windowed fragment with from os.path import join as j keeps
  its PE3 coverage under a scan

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 24 ++++++++++----
 .../nodes/analyzers/test_shared_python_ast.py |  7 +++--
 tests/unit/test_patterns.py                   | 31 +++++++++++++++++++
 3 files changed, 54 insertions(+), 8 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 98cd32712..c25f0ed45 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -683,6 +683,14 @@ def _is_qualified_benign_access_requirement(
 _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.
@@ -716,8 +724,8 @@ def _constructed_sensitive_paths(
     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 and are parsed directly behind the same
-    textual join gate, so large files keep their findings instead of silently
-    dropping them.  Only fully-literal positional argument lists are resolved;
+    textual join-or-import gate, so large files keep their findings instead of
+    silently dropping them.  Only fully-literal positional argument lists are resolved;
     anything dynamic is left to the existing pattern loop.  Unparseable content
     simply yields no findings here.
     """
@@ -725,13 +733,17 @@ def _constructed_sensitive_paths(
         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 and _join_call_hint({}).search(content):
+            if python_ast is None and (
+                _join_call_hint({}).search(content) or _JOIN_IMPORT_HINT.search(content)
+            ):
                 # A windowed view fragment, not the scan's whole file: the
                 # shared tree does not cover this slice, so parse the
                 # fragment directly to avoid silently dropping large-file
-                # findings.  The plain textual gate keeps the runner's
-                # scan-count discipline for fragments without a plausible
-                # join call.
+                # findings.  The textual gate also fires on ``from os.path
+                # import`` spellings so renamed calls (``from os.path import
+                # join as j`` then ``j(``) survive; the alias-aware gate
+                # below filters import-only fragments.  Fragments without a
+                # plausible join call or join import never parse.
                 python_ast = parse_python_source(content, file_path)
         else:
             python_ast = parse_python_source(content, file_path)
diff --git a/tests/nodes/analyzers/test_shared_python_ast.py b/tests/nodes/analyzers/test_shared_python_ast.py
index 7ff47d8b0..052a52cfe 100644
--- a/tests/nodes/analyzers/test_shared_python_ast.py
+++ b/tests/nodes/analyzers/test_shared_python_ast.py
@@ -173,10 +173,13 @@ def test_graph_scan_reports_constructed_path_above_view_window_chars(tmp_path) -
     ``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 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 = "import os\ncredential = os.path.join('/etc', 'passwd')\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
diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py
index 612c0c1e0..1c6cd1701 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -405,6 +405,37 @@ def test_pe3_windowed_fragment_keeps_constructed_path_coverage(self) -> None:
             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_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"

From fb9d631c11550731be70e3155dffa9457bd6c5ec Mon Sep 17 00:00:00 2001
From: deepujain <deepujain@gmail.com>
Date: Wed, 23 Sep 2026 02:32:35 +0000
Subject: [PATCH 8/9] fix(patterns): carry whole-file join aliases into
 windowed fragments

A renamed os.path.join import (from os.path import join as j) and its call
site can fall in different runner windows. The call's fragment carries no
import aliases of its own and never spells join(, so the textual gate
skipped it before alias resolution and the constructed sensitive path was
silently dropped.

Carry the whole file's import-alias map from the scan cache into the
fragment analysis: peek the whole-file parse by filename (no content
match), use its aliases for the join pre-check gate, the alias-aware gate,
and call resolution. Fragments of the same scanned file share its import
bindings, so the renamed call resolves to os.path.join in the later
window. Standalone and exact-cache-hit paths are unchanged, and fragments
without a plausible join call or join import still never parse.

Adds two regression tests: an analyze-level fragment whose import lives in
another window, and a full node() graph regression with import and call
separated across real runner windows.

Signed-off-by: deepujain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 40 ++++++++----
 src/skillspector/python_ast.py                | 19 ++++++
 tests/unit/test_patterns.py                   | 61 +++++++++++++++++++
 3 files changed, 107 insertions(+), 13 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index c25f0ed45..528c4a6da 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -26,7 +26,12 @@
 
 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
+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
@@ -725,26 +730,35 @@ def _constructed_sensitive_paths(
     callers get a single on-demand parse.  Windowed view fragments under a scan
     miss the whole-file cache entry and are parsed directly behind the same
     textual join-or-import gate, so large files keep their findings instead of
-    silently dropping them.  Only fully-literal positional argument lists are resolved;
+    silently dropping them.  On such a fragment 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.  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
     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 and (
-                _join_call_hint({}).search(content) or _JOIN_IMPORT_HINT.search(content)
-            ):
+            if python_ast is None:
                 # A windowed view fragment, not the scan's whole file: the
-                # shared tree does not cover this slice, so parse the
-                # fragment directly to avoid silently dropping large-file
-                # findings.  The textual gate also fires on ``from os.path
-                # import`` spellings so renamed calls (``from os.path import
-                # join as j`` then ``j(``) survive; the alias-aware gate
-                # below filters import-only fragments.  Fragments without a
+                # shared tree does not cover this slice.  Carry the whole
+                # file's import-alias map 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.
-                python_ast = parse_python_source(content, file_path)
+                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
+                if _join_call_hint(whole_file_aliases or {}).search(
+                    content
+                ) or _JOIN_IMPORT_HINT.search(content):
+                    python_ast = parse_python_source(content, file_path)
         else:
             python_ast = parse_python_source(content, file_path)
     if python_ast is None:
@@ -752,7 +766,7 @@ def _constructed_sensitive_paths(
     tree = python_ast.tree
     if tree is None:
         return []
-    aliases = python_ast.import_aliases
+    aliases = whole_file_aliases if whole_file_aliases is not None else python_ast.import_aliases
     if not _join_call_hint(aliases).search(content):
         return []
     resolved: list[tuple[int, int, str, float]] = []
diff --git a/src/skillspector/python_ast.py b/src/skillspector/python_ast.py
index a26a3c4c0..2637615a2 100644
--- a/src/skillspector/python_ast.py
+++ b/src/skillspector/python_ast.py
@@ -358,6 +358,25 @@ def peek_python_ast(cache_key: str | None, content: str, filename: str) -> Parse
         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/unit/test_patterns.py b/tests/unit/test_patterns.py
index 1c6cd1701..d9ddc5c90 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -436,6 +436,67 @@ def test_pe3_windowed_fragment_with_renamed_join_import_keeps_coverage(self) ->
             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_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"

From d1f9339ddbda7b60304311b8673fe9b8e621ab3c Mon Sep 17 00:00:00 2001
From: Deepak Jain <deepujain@gmail.com>
Date: Wed, 23 Sep 2026 03:31:27 +0000
Subject: [PATCH 9/9] fix(patterns): evaluate windowed join calls on the
 whole-file AST

The windowed fallback parsed each fragment as a standalone module, so a
slice starting mid-block (for example inside a function body) failed
parsing and its constructed-path PE3 findings were silently dropped.
Evaluate the supplemental constructed calls against the whole file's
cached tree instead, reusing the existing cache peek: call spans map
onto fragment-relative lines and only calls starting inside the
fragment are owned by it, preserving full-file syntax, scope and
import aliases as well as the runner's source-coordinate and dedupe
behavior. Fragments that are not slices of the whole file (normalized
views) keep the standalone-parse fallback.

Add direct and graph regressions using the enclosing-function fixture:
a mid-function window now reports exactly one PE3 at the call site.

Signed-off-by: Deepak Jain <deepujain@gmail.com>
---
 .../static_patterns_privilege_escalation.py   | 89 +++++++++++++++----
 tests/unit/test_patterns.py                   | 71 +++++++++++++++
 2 files changed, 144 insertions(+), 16 deletions(-)

diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
index 528c4a6da..d3fdd26ba 100644
--- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
+++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py
@@ -712,6 +712,21 @@ def _join_call_hint(aliases: dict[str, str]) -> re.Pattern[str]:
     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,
@@ -728,17 +743,27 @@ def _constructed_sensitive_paths(
     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 and are parsed directly behind the same
-    textual join-or-import gate, so large files keep their findings instead of
-    silently dropping them.  On such a fragment the whole file's import-alias
+    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.  Only fully-literal positional argument lists are resolved;
-    anything dynamic is left to the existing pattern loop.  Unparseable content
-    simply yields no findings here.
+    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:
@@ -746,8 +771,8 @@ def _constructed_sensitive_paths(
             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 from the scan cache first (a
-                # fragment is a slice of the same scanned file), so the
+                # 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
@@ -755,18 +780,40 @@ def _constructed_sensitive_paths(
                 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):
-                    python_ast = parse_python_source(content, file_path)
+                    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 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
+    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]] = []
@@ -781,9 +828,19 @@ def _constructed_sensitive_paths(
         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((node.lineno, node.end_lineno or node.lineno, value, confidence))
+                resolved.append((start_line, end_line, value, confidence))
                 break
     return resolved
 
diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py
index d9ddc5c90..6080d1eee 100644
--- a/tests/unit/test_patterns.py
+++ b/tests/unit/test_patterns.py
@@ -497,6 +497,77 @@ def test_pe3_graph_windowed_import_and_call_across_windows(self) -> None:
         ]
         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"