fix(mcp): address review — add tests, methodology refs for new tools

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) <noreply@anthropic.com>

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

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]

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