From dded7d01f45076c32185c2692206c251f73cbc38 Mon Sep 17 00:00:00 2001 From: CharmingGroot Date: Wed, 24 Jun 2026 15:51:54 +0900 Subject: [PATCH 1/2] feat(analyzer): detect Docker socket access as PE4 privilege escalation Add PE4_PATTERNS to the static privilege escalation analyzer to detect Docker socket access, which allows a containerized skill to gain full host root without triggering any of the existing PE1-PE3 rules. Four patterns are added: - /var/run/docker.sock (socket path, confidence 0.9) - docker.from_env() (SDK entry point, confidence 0.85) - DockerClient( (explicit client construction, confidence 0.85) - http+unix://...docker.sock (raw HTTP-over-socket, confidence 0.9) Closes #188 Signed-off-by: CharmingGroot --- .../static_patterns_privilege_escalation.py | 28 +++++- tests/nodes/analyzers/test_static_patterns.py | 96 +++++++++++++++++++ 2 files changed, 122 insertions(+), 2 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py index 3a7661c7..385592ed 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,24 @@ def loc(ln: int) -> Location: matched_text=match.group(0)[:200], ) ) + 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 + findings.append( + 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], + ) + ) return findings diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index 799f245b..96266d11 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -328,3 +328,99 @@ def test_node_runs_over_state(self): } result = agent_snooping_module.node(state) 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_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\n" + "For example:\n" + "```python\n" + "client = 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"]) From 34309b324d6950317d98d7c8273982c732cc1d6c Mon Sep 17 00:00:00 2001 From: CharmingGroot Date: Wed, 24 Jun 2026 17:41:30 +0900 Subject: [PATCH 2/2] fix(analyzer): deduplicate PE4 findings per line to avoid double-reporting A line such as `docker.DockerClient(base_url="unix:///var/run/docker.sock")` matched both the `DockerClient(` and `/var/run/docker.sock` PE4 patterns, producing two findings for the same vulnerability on the same line. Collect PE4 candidates in a dict keyed by line number, keeping only the highest-confidence match, then extend the findings list once. Adds a regression test that asserts exactly one PE4 finding on such a combined line. Addresses nit raised in review of PR #189. Signed-off-by: CharmingGroot Signed-off-by: CharmingGroot --- .../static_patterns_privilege_escalation.py | 26 +++++++++++-------- tests/nodes/analyzers/test_static_patterns.py | 23 ++++++++++++---- 2 files changed, 33 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 385592ed..e8742488 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py +++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py @@ -162,24 +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 - findings.append( - 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], - ) + 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 574f9a6c..b0e3454c 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -354,6 +354,23 @@ def test_pe4_docker_sock_path_produces_finding(self): 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 = { @@ -406,11 +423,7 @@ def test_pe4_documentation_example_not_flagged(self): "components": ["SKILL.md"], "file_cache": { "SKILL.md": ( - "# Docker SDK\n\n" - "For example:\n" - "```python\n" - "client = docker.from_env()\n" - "```\n" + "# Docker SDK\n\nFor example:\n```python\nclient = docker.from_env()\n```\n" ), }, }