feat(analyzer): detect Docker socket access as PE4 privilege escalation

src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Static patterns: privilege escalation (PE1–PE3). Node and analyze() in one module."""
+"""Static patterns: privilege escalation (PE1–PE4). Node and analyze() in one module."""
 
 from __future__ import annotations
 
@@ -93,10 +93,16 @@
     (r"access\s+(?:the\s+)?(?:credentials?|secrets?|tokens?)", 0.7),
     (r"(?:extract|copy|get)\s+(?:api\s+)?keys?\s+from", 0.7),
 ]
+PE4_PATTERNS = [
+    (r"/var/run/docker\.sock", 0.9),
+    (r"docker\.from_env\(\)", 0.85),
+    (r"\bDockerClient\s*\(", 0.85),
+    (r"http\+unix://.*docker\.sock", 0.9),
+]
 
 
 def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]:
-    """Analyze content for privilege escalation patterns (PE1–PE3)."""
+    """Analyze content for privilege escalation patterns (PE1–PE4)."""
     findings: list[AnalyzerFinding] = []
 
     def loc(ln: int) -> Location:
@@ -156,6 +162,28 @@ def loc(ln: int) -> Location:
                     matched_text=match.group(0)[:200],
                 )
             )
+    # 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] = {}
+    for pattern, confidence in PE4_PATTERNS:
+        for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE):
+            line_num = get_line_number(content, match.start())
+            context = get_context(content, match.start())
+            if _is_documentation_example(context, file_type):
+                continue
+            if line_num in pe4_best and pe4_best[line_num].confidence >= confidence:
+                continue
+            pe4_best[line_num] = AnalyzerFinding(
+                rule_id="PE4",
+                message="Docker Socket Access",
+                severity=Severity.HIGH,
+                location=loc(line_num),
+                confidence=confidence,
+                tags=tag,
+                context=context,
+                matched_text=match.group(0)[:200],
+            )
+    findings.extend(pe4_best.values())
     return findings
 
 

tests/nodes/analyzers/test_static_patterns.py

@@ -333,6 +333,115 @@ def test_node_runs_over_state(self):
         assert any(f.rule_id == "AS1" for f in result["findings"])
 
 
+class TestRunStaticPatternsPrivilegeEscalationPE4:
+    """run_static_patterns with privilege_escalation: PE4 (Docker socket access)."""
+
+    def test_pe4_docker_sock_path_produces_finding(self):
+        """Direct reference to /var/run/docker.sock yields PE4 (HIGH)."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": 'client = docker.DockerClient(base_url="unix:///var/run/docker.sock")\n',
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        pe4 = [f for f in findings if f.rule_id == "PE4"]
+        assert len(pe4) >= 1
+        assert pe4[0].severity == "HIGH"
+        assert pe4[0].file == "skill.py"
+        assert pe4[0].start_line >= 1
+        assert pe4[0].remediation is not None
+        assert pe4[0].context is not None
+        assert pe4[0].matched_text is not None
+
+    def test_pe4_combined_line_produces_exactly_one_finding(self):
+        """A line matching multiple PE4 patterns must produce exactly one PE4 finding."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": 'client = docker.DockerClient(base_url="unix:///var/run/docker.sock")\n',
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        pe4 = [f for f in findings if f.rule_id == "PE4"]
+        assert len(pe4) == 1, (
+            f"Expected 1 PE4 finding, got {len(pe4)}: {[f.matched_text for f in pe4]}"
+        )
+        assert (
+            pe4[0].confidence == 0.9
+        )  # /var/run/docker.sock has higher confidence than DockerClient(
+
+    def test_pe4_docker_from_env_produces_finding(self):
+        """docker.from_env() yields PE4 (HIGH)."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": "import docker\nclient = docker.from_env()\n",
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        pe4 = [f for f in findings if f.rule_id == "PE4"]
+        assert len(pe4) >= 1
+        assert pe4[0].severity == "HIGH"
+
+    def test_pe4_docker_client_constructor_produces_finding(self):
+        """DockerClient( instantiation yields PE4 (HIGH)."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": "from docker import DockerClient\nclient = DockerClient(base_url='tcp://...')\n",
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        assert any(f.rule_id == "PE4" for f in findings)
+
+    def test_pe4_http_unix_socket_produces_finding(self):
+        """http+unix:// reference to docker.sock yields PE4 (HIGH)."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": 'url = "http+unix://%2Fvar%2Frun%2Fdocker.sock/containers/json"\n',
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        assert any(f.rule_id == "PE4" for f in findings)
+
+    def test_pe4_safe_docker_subprocess_not_flagged(self):
+        """subprocess call to docker CLI without socket reference produces no PE4."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": "subprocess.run(['docker', 'ps', '--format', 'json'])\n",
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        assert not any(f.rule_id == "PE4" for f in findings)
+
+    def test_pe4_documentation_example_not_flagged(self):
+        """docker.from_env() inside a markdown code block is filtered as documentation."""
+        state = {
+            "components": ["SKILL.md"],
+            "file_cache": {
+                "SKILL.md": (
+                    "# Docker SDK\n\nFor example:\n```python\nclient = docker.from_env()\n```\n"
+                ),
+            },
+        }
+        findings = static_runner.run_static_patterns(state, [privilege_escalation_module])
+        assert not any(f.rule_id == "PE4" for f in findings)
+
+    def test_pe4_node_runs_over_state(self):
+        """The node entrypoint runs PE4 detection over state and returns findings."""
+        state = {
+            "components": ["skill.py"],
+            "file_cache": {
+                "skill.py": "client = docker.from_env()\n",
+            },
+        }
+        result = privilege_escalation_module.node(state)
+        assert any(f.rule_id == "PE4" for f in result["findings"])
+
+
 class TestRunStaticPatternsSSRF:
     """run_static_patterns with ssrf: SSRF1, SSRF2, SSRF3."""