From f239412bbd5911309443d4fc699de4abe3a39be3 Mon Sep 17 00:00:00 2001 From: Ms6RB Date: Thu, 26 Mar 2026 01:19:07 +0200 Subject: [PATCH] =?UTF-8?q?fix(mcp):=20address=20review=20=E2=80=94=20add?= =?UTF-8?q?=20tests,=20methodology=20refs=20for=20new=20tools?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add k8s_enumerate tests (4), ssrf_oracle tests (2), body_format_warning tests (2). Add k8s_enumerate, ssrf_oracle, oauth_audit, webhook_ssrf, dangling_resources, pg_tenant_audit to methodology recon directives. Co-Authored-By: Claude Opus 4.6 (1M context) --- strix-mcp/src/strix_mcp/methodology.md | 5 ++ strix-mcp/tests/test_chaining.py | 21 +++++ strix-mcp/tests/test_tools_analysis.py | 101 +++++++++++++++++++++++++ 3 files changed, 127 insertions(+) diff --git a/strix-mcp/src/strix_mcp/methodology.md b/strix-mcp/src/strix_mcp/methodology.md index 3655cc38..764c8485 100644 --- a/strix-mcp/src/strix_mcp/methodology.md +++ b/strix-mcp/src/strix_mcp/methodology.md @@ -125,6 +125,11 @@ Before vulnerability testing, run reconnaissance to map the full attack surface. - If SAML/SSO endpoints detected → dispatch SSO agent with `load_skill("saml_sso_bypass")` - Run `test_request_smuggling` when target is behind a CDN or reverse proxy — detects CL.TE/TE.CL/TE.0 parser discrepancies - Run `test_cache_poisoning` when target uses caching (CDN detected) — finds unkeyed headers and cache deception vectors +- If OAuth server detected → dispatch agent with `load_skill("oauth_audit")` for systematic client enumeration, redirect_uri DNS checks, and PKCE testing +- If webhooks/callbacks found → dispatch agent with `load_skill("webhook_ssrf")` for systematic SSRF bypass testing +- After confirming blind SSRF → use `ssrf_oracle` to calibrate retry/timing/status oracles, then use `k8s_enumerate` to generate internal service wordlists for probing +- If target uses managed PostgreSQL (Neon, Supabase, etc.) → dispatch agent with `load_skill("pg_tenant_audit")` +- Run `load_skill("dangling_resources")` to check all external references (OAuth redirect_uris, CNAMEs, integrations) for NXDOMAIN/expired domains - Load skill `browser_security` when testing custom browsers (Electron, Chromium forks) or AI-powered browsers — contains address bar spoofing test templates, prompt injection vectors, and UI spoofing detection methodology - Write ALL results as structured notes: `create_note(category="recon", title="...")` - Stay within scope: check `scope_rules` before scanning new targets diff --git a/strix-mcp/tests/test_chaining.py b/strix-mcp/tests/test_chaining.py index f3a68550..35830c61 100644 --- a/strix-mcp/tests/test_chaining.py +++ b/strix-mcp/tests/test_chaining.py @@ -481,3 +481,24 @@ class TestReasonCrossToolChains: assert "next_action" in chain assert isinstance(chain["evidence"], list) assert isinstance(chain["missing"], list) + + def test_ssrf_webhook_body_format_warning(self): + """SSRF chain with webhook in title should include body_format_warning.""" + js = {"internal_hostnames": ["https://10.0.1.50:8080"], "collection_names": [], "secrets": []} + vulns = [{"title": "Webhook SSRF in /api/hooks", "severity": "high"}] + + chains = reason_cross_tool_chains(js_analysis=js, vuln_reports=vulns) + ssrf_chains = [c for c in chains if "SSRF" in c["name"]] + assert len(ssrf_chains) >= 1 + assert "body_format_warning" in ssrf_chains[0] + assert "redirect" in ssrf_chains[0]["body_format_warning"].lower() + + def test_ssrf_no_webhook_no_body_warning(self): + """SSRF chain without webhook should NOT include body_format_warning.""" + js = {"internal_hostnames": ["https://10.0.1.50:8080"], "collection_names": [], "secrets": []} + vulns = [{"title": "SSRF in image proxy", "severity": "high"}] + + chains = reason_cross_tool_chains(js_analysis=js, vuln_reports=vulns) + ssrf_chains = [c for c in chains if "SSRF" in c["name"]] + assert len(ssrf_chains) >= 1 + assert "body_format_warning" not in ssrf_chains[0] diff --git a/strix-mcp/tests/test_tools_analysis.py b/strix-mcp/tests/test_tools_analysis.py index 08b1a53a..24df8c55 100644 --- a/strix-mcp/tests/test_tools_analysis.py +++ b/strix-mcp/tests/test_tools_analysis.py @@ -1250,3 +1250,104 @@ class TestCachePoisoning: assert result["cache_detected"] is True assert result["cache_type"] == "cloudflare" + + +class TestK8sEnumerate: + """Tests for the k8s_enumerate MCP tool.""" + + @pytest.fixture + def mcp_k8s(self): + mcp = FastMCP("test-strix") + mock_sandbox = MagicMock() + mock_sandbox.active_scan = None + mock_sandbox._active_scan = None + register_tools(mcp, mock_sandbox) + return mcp + + @pytest.mark.asyncio + async def test_default_wordlist(self, mcp_k8s): + result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {}))) + assert result["total_urls"] > 100 + assert "urls_by_namespace" in result + assert "kube-system" in result["urls_by_namespace"] + + @pytest.mark.asyncio + async def test_target_name_adds_custom_services(self, mcp_k8s): + result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", { + "target_name": "neon", + }))) + all_urls = [] + for ns_urls in result["urls_by_namespace"].values(): + all_urls.extend(ns_urls) + assert any("neon-api" in u for u in all_urls) + assert "neon" in result["urls_by_namespace"] # namespace added + + @pytest.mark.asyncio + async def test_custom_namespaces_and_ports(self, mcp_k8s): + result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", { + "namespaces": ["custom-ns"], + "ports": [9999], + }))) + assert "custom-ns" in result["urls_by_namespace"] + all_urls = [] + for ns_urls in result["urls_by_namespace"].values(): + all_urls.extend(ns_urls) + assert any(":9999" in u for u in all_urls) + + @pytest.mark.asyncio + async def test_result_structure(self, mcp_k8s): + result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {}))) + for key in ["total_urls", "urls_by_namespace", "short_forms", "usage_hint"]: + assert key in result + assert isinstance(result["short_forms"], list) + + +class TestSsrfOracle: + """Tests for the ssrf_oracle MCP tool.""" + + @pytest.fixture + def mcp_no_scan(self): + mcp = FastMCP("test-strix") + mock_sandbox = MagicMock() + mock_sandbox.active_scan = None + mock_sandbox._active_scan = None + register_tools(mcp, mock_sandbox) + return mcp + + @pytest.fixture + def mcp_with_scan(self): + from unittest.mock import AsyncMock + mcp = FastMCP("test-strix") + mock_sandbox = MagicMock() + scan = ScanState( + scan_id="test", + workspace_id="ws-1", + api_url="http://localhost:8080", + token="tok", + port=8080, + default_agent_id="mcp-test", + ) + mock_sandbox.active_scan = scan + mock_sandbox._active_scan = scan + mock_sandbox.proxy_tool = AsyncMock(return_value={ + "response": {"status_code": 200, "body": "ok"}, + }) + register_tools(mcp, mock_sandbox) + return mcp, mock_sandbox + + @pytest.mark.asyncio + async def test_no_active_scan(self, mcp_no_scan): + result = json.loads(_tool_text(await mcp_no_scan.call_tool("ssrf_oracle", { + "ssrf_url": "https://target.com/webhook", + }))) + assert "error" in result + + @pytest.mark.asyncio + async def test_result_structure(self, mcp_with_scan): + mcp, _ = mcp_with_scan + result = json.loads(_tool_text(await mcp.call_tool("ssrf_oracle", { + "ssrf_url": "https://target.com/webhook", + }))) + assert "oracles" in result + assert "baseline" in result + assert "recommended_approach" in result