diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py index 3a7661c7..e8742488 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py +++ b/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 diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index e5dd1e95..b0e3454c 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/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."""