scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-09-11 22:51:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(mcp): k8s_enumerate namespace affinity, ssrf_oracle https probes

Browse source 
- k8s_enumerate: services mapped to likely namespaces (grafana→monitoring,
  kubernetes→default, argocd-server→argocd, etc). Unmapped services only
  in default+kube-system. Reduces 488→73 URLs. max_urls=0 returns empty.
- ssrf_oracle: use https:// for all test URLs to isolate IP/hostname
  validation from scheme validation. Document retry oracle limitation.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ms6RB 2026-03-26 01:59:37 +02:00
parent 94e4e997d3
commit 970cf82e09
2 changed files with 72 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
strix-mcp/src/strix_mcp/tools_analysis.py
View file

@ -1293,25 +1293,60 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
default_namespaces.append(target_name.lower().strip())
ns_list = namespaces or default_namespaces
# Generate URLs grouped by namespace (service-specific ports, not cartesian)
by_namespace: dict[str, list[str]] = {}
# Namespace affinity: map services to their likely namespaces
# Only generate URLs for plausible service→namespace combinations
ns_affinity: dict[str, list[str]] = {
"default": ["kubernetes"],
"kube-system": ["kube-dns", "coredns", "metrics-server", "aws-load-balancer-controller",
"external-dns", "ebs-csi-controller", "cluster-autoscaler"],
"monitoring": ["grafana", "prometheus", "alertmanager", "victoria-metrics",
"thanos", "loki", "tempo"],
"argocd": ["argocd-server"],
"vault": ["vault"],
"cert-manager": ["cert-manager"],
"istio-system": ["istiod", "istio-ingressgateway", "envoy", "linkerd-controller"],
}
# Target-specific services go to target namespace
if target_name:
name = target_name.lower().strip()
ns_affinity[name] = [f"{name}{s}" for s in ["-api", "-proxy", "-auth", "-control-plane", "-storage", "-compute"]]
# Services not in any affinity map go to all namespaces
mapped_services = set()
for svcs in ns_affinity.values():
mapped_services.update(svcs)
unmapped = [s for s in service_ports if s not in mapped_services]
# Generate URLs — use affinity when available, fallback to default+kube-system for unmapped
by_namespace: dict[str, list[str]] = {ns: [] for ns in ns_list}
total = 0
for ns in ns_list:
urls: list[str] = []
for svc, svc_ports in service_ports.items():
for port in svc_ports:
urls.append(f"{scheme}://{svc}.{ns}.svc.cluster.local:{port}")
total += 1
by_namespace[ns] = urls
affinity_svcs = ns_affinity.get(ns, [])
for svc in affinity_svcs:
if svc in service_ports:
for port in service_ports[svc]:
by_namespace[ns].append(f"{scheme}://{svc}.{ns}.svc.cluster.local:{port}")
total += 1
# Unmapped services only go to default and kube-system
if ns in ("default", "kube-system"):
for svc in unmapped:
for port in service_ports[svc]:
by_namespace[ns].append(f"{scheme}://{svc}.{ns}.svc.cluster.local:{port}")
total += 1
# Also generate short-form names (service only, no namespace cross-product)
short_forms: list[str] = []
for svc in service_ports:
short_forms.append(f"{scheme}://{svc}")
# Remove empty namespaces
by_namespace = {ns: urls for ns, urls in by_namespace.items() if urls}
# Short-form names (service only)
short_forms: list[str] = [f"{scheme}://{svc}" for svc in service_ports]
# Cap output — distribute evenly across namespaces
omitted = 0
if total > max_urls:
if max_urls <= 0:
by_namespace = {ns: [] for ns in by_namespace}
omitted = total
total = 0
elif total > max_urls:
per_ns = max(max_urls // len(by_namespace), 1)
new_total = 0
for ns in by_namespace:
@ -1354,18 +1389,26 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
Requires an active sandbox.
Given a confirmed blind SSRF endpoint, tests with known-good and known-bad
targets to build an oracle (retry behavior, timing, status codes) that can
distinguish successful from failed internal requests.
targets to build an oracle (timing, status codes) that can distinguish
successful from failed internal requests.
ssrf_url: the vulnerable endpoint URL
ssrf_url: the vulnerable endpoint URL (e.g. webhook config endpoint)
ssrf_param: parameter name that accepts the target URL (default "url")
ssrf_method: HTTP method (default POST)
ssrf_headers: additional headers for the SSRF request
ssrf_body_template: request body template with {TARGET_URL} placeholder
agent_id: subagent identifier from dispatch_agent
Returns: oracle calibration data — baseline responses, retry behavior,
timing differentials, and recommended exploitation approach."""
NOTE on retry oracle: This tool detects timing and status differentials
from the SSRF config endpoint response. For webhook-style SSRFs where the
real oracle is in delivery retries, you need a 2-phase approach:
(1) set webhook URL to a redirect → interactsh/webhook.site
(2) trigger the event that fires the webhook
(3) count incoming requests at the receiver
Use python_action for this — this tool handles the config-response oracle.
Returns: oracle calibration data — baseline responses, timing differentials,
and recommended exploitation approach."""
scan = sandbox.active_scan
if scan is None:
@ -1430,8 +1473,8 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
# --- Phase 1: Baseline calibration ---
probe_targets = {
"reachable": "https://httpbin.org/status/200",
"unreachable": "http://192.0.2.1/",
"dns_fail": "http://this-domain-does-not-exist-strix-test.invalid/",
"unreachable": "https://192.0.2.1/",
"dns_fail": "https://this-domain-does-not-exist-strix-test.invalid/",
}
baseline: dict[str, Any] = {}

17
strix-mcp/tests/test_tools_analysis.py
View file

@ -1267,8 +1267,8 @@ class TestK8sEnumerate:
@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"] > 50
assert result["total_urls"] < 500 # no longer a cartesian product
assert result["total_urls"] > 20 # affinity reduces count
assert result["total_urls"] < 500
assert "urls_by_namespace" in result
assert "kube-system" in result["urls_by_namespace"]
@ -1294,12 +1294,12 @@ class TestK8sEnumerate:
@pytest.mark.asyncio
async def test_service_specific_ports(self, mcp_k8s):
"""Services should use their known default ports, not a cartesian product."""
# grafana has affinity to 'monitoring' namespace, so test there
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"namespaces": ["default"],
"namespaces": ["monitoring"],
})))
urls = result["urls_by_namespace"]["default"]
# grafana should only appear on port 3000 (its default), not on 443, 6379, etc.
grafana_urls = [u for u in urls if "grafana.default" in u]
urls = result["urls_by_namespace"].get("monitoring", [])
grafana_urls = [u for u in urls if "grafana.monitoring" in u]
grafana_ports = [int(u.split(":")[-1]) for u in grafana_urls]
assert 3000 in grafana_ports
assert 6379 not in grafana_ports # redis port should not be on grafana
@ -1318,11 +1318,12 @@ class TestK8sEnumerate:
@pytest.mark.asyncio
async def test_additional_ports_appended(self, mcp_k8s):
"""User-supplied ports should be added to service defaults, not replace them."""
# Use monitoring namespace where grafana has affinity
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"namespaces": ["default"],
"namespaces": ["monitoring"],
"ports": [9999],
})))
urls = result["urls_by_namespace"]["default"]
urls = result["urls_by_namespace"].get("monitoring", [])
# 9999 should appear as additional port on services
assert any(":9999" in u for u in urls)
# grafana's default 3000 should still be present