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

fix(mcp): fix download_sourcemaps module scripts, k8s_enumerate output, load_skill overflow

Browse source 
- download_sourcemaps: fix regex to match type=module crossorigin scripts
- k8s_enumerate: map services to default ports instead of cartesian product,
  add scheme parameter (default https), cap output size
- load_skill: add max_content_length (50K) and summary_only mode to prevent
  MCP buffer overflow on large skills

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ms6RB 2026-03-26 01:29:06 +02:00
parent f239412bbd
commit 0e4e26037f
7 changed files with 255 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
strix-mcp/src/strix_mcp/tools.py
View file

@ -424,13 +424,21 @@ def register_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
return resources.list_modules(category=category)
@mcp.tool()
async def load_skill(skills: str) -> str:
async def load_skill(
skills: str,
max_content_length: int = 50000,
summary_only: bool = False,
) -> str:
"""Dynamically load security knowledge skills into the current conversation.
Runs client-side (no sandbox required). Returns the full skill content
inline so you can immediately apply the techniques described.
skills: comma-separated skill names (max 5). Use list_modules to see
available skills. Examples: "nuclei,sqlmap", "xss", "graphql,nextjs"
max_content_length: maximum total chars for all skill content (default 50000).
If exceeded, the largest skills are truncated with a note.
summary_only: if True, return just skill names and descriptions without
full content (useful for checking what would be loaded)
Prefer this over get_module when you need to actively apply multiple skills
at once. The returned content includes exploitation techniques, tool usage,
@ -479,7 +487,44 @@ def register_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
}
if failed:
result["failed_skills"] = failed
result["skill_content"] = loaded_content
if summary_only:
# Return just names and first-line descriptions
result["skill_summaries"] = {
name: content.split("\n", 1)[0][:200]
for name, content in loaded_content.items()
}
else:
# Apply max_content_length: truncate largest skills first
total_len = sum(len(c) for c in loaded_content.values())
if total_len > max_content_length:
# Sort by size descending to truncate largest first
by_size = sorted(loaded_content.items(), key=lambda x: -len(x[1]))
truncated_content: dict[str, str] = {}
truncation_notes: list[str] = []
remaining_budget = max_content_length
# First pass: calculate fair share per skill
for name, content in sorted(loaded_content.items(), key=lambda x: len(x[1])):
skills_left = len(loaded_content) - len(truncated_content)
fair_share = remaining_budget // max(skills_left, 1)
if len(content) <= fair_share:
truncated_content[name] = content
remaining_budget -= len(content)
else:
limit = max(fair_share, 500) # keep at least 500 chars
truncated_content[name] = content[:limit]
truncation_notes.append(
f"Skill '{name}' truncated to {limit} chars. "
"Call load_skill with fewer skills to get full content."
)
remaining_budget -= limit
result["skill_content"] = truncated_content
if truncation_notes:
result["truncation_notes"] = truncation_notes
else:
result["skill_content"] = loaded_content
return json.dumps(result)

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

@ -1201,67 +1201,88 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
# --- K8s Service Enumeration Wordlist Generator ---
# Service registry: maps service name -> default ports
K8S_SERVICES: dict[str, list[int]] = {
# K8s core
"kubernetes": [443, 6443],
"kube-dns": [53],
"metrics-server": [443],
"coredns": [53],
# Monitoring
"grafana": [3000],
"prometheus": [9090],
"alertmanager": [9093],
"victoria-metrics": [8428],
"thanos": [9090, 10901],
"loki": [3100],
"tempo": [3200],
# GitOps
"argocd-server": [443, 8080],
# Security
"vault": [8200],
"cert-manager": [9402],
# Service mesh
"istiod": [15010, 15012],
"istio-ingressgateway": [443, 80],
# Auth
"keycloak": [8080, 8443],
"hydra": [4444, 4445],
"dex": [5556],
"oauth2-proxy": [4180],
# Data
"redis": [6379],
"rabbitmq": [5672, 15672],
"kafka": [9092],
"elasticsearch": [9200],
"nats": [4222],
# AWS EKS
"aws-load-balancer-controller": [9443],
"external-dns": [7979],
"ebs-csi-controller": [9808],
"cluster-autoscaler": [8085],
}
_TARGET_DEFAULT_PORTS = [443, 8080, 5432, 3000]
@mcp.tool()
async def k8s_enumerate(
target_name: str | None = None,
namespaces: list[str] | None = None,
ports: list[int] | None = None,
scheme: str = "https",
max_urls: int = 500,
) -> str:
"""Generate a comprehensive K8s service enumeration wordlist for SSRF probing.
"""Generate a K8s service enumeration wordlist for SSRF probing.
No sandbox required.
Returns service URLs to test via SSRF. Feed these into send_request,
python_action, or the webhook URL parameter to discover internal services.
Returns service URLs to test via SSRF. Each service is mapped to its
known default ports (not a cartesian product), keeping the list compact.
target_name: company/product name for generating custom service names (e.g. "neon")
namespaces: custom namespaces (default: common K8s namespaces)
ports: custom ports (default: common service ports)
ports: ADDITIONAL ports to scan on top of each service's defaults
scheme: URL scheme (default "https")
max_urls: maximum URLs to return (default 500)
Usage: get the URL list, then use python_action to spray them through
your SSRF vector and observe which ones resolve."""
# Standard K8s services
services = [
"kubernetes", "kube-dns", "metrics-server", "coredns",
]
# AWS EKS
services += [
"aws-load-balancer-controller", "external-dns",
"ebs-csi-controller", "cluster-autoscaler",
]
# Monitoring
services += [
"grafana", "prometheus", "alertmanager", "victoria-metrics",
"thanos", "loki", "tempo",
]
# GitOps
services += [
"argocd-server", "flux-source-controller", "flux-helm-controller",
]
# Security
services += [
"vault", "cert-manager", "falco", "trivy-operator",
]
# Service mesh
services += [
"istiod", "istio-ingressgateway", "envoy", "linkerd-controller",
]
# Auth
services += [
"keycloak", "hydra", "dex", "oauth2-proxy",
]
# Data
services += [
"redis", "rabbitmq", "kafka", "elasticsearch", "nats",
]
# Build service -> ports mapping (start from registry defaults)
service_ports: dict[str, list[int]] = {
svc: list(svc_ports) for svc, svc_ports in K8S_SERVICES.items()
}
# Target-specific services
# Target-specific services with default ports
if target_name:
name = target_name.lower().strip()
services += [
f"{name}-api", f"{name}-proxy", f"{name}-auth",
f"{name}-control-plane", f"{name}-storage", f"{name}-compute",
]
for suffix in ["-api", "-proxy", "-auth", "-control-plane", "-storage", "-compute"]:
service_ports[f"{name}{suffix}"] = list(_TARGET_DEFAULT_PORTS)
# Append user-supplied additional ports to every service
if ports:
for svc in service_ports:
for p in ports:
if p not in service_ports[svc]:
service_ports[svc].append(p)
# Namespaces
default_namespaces = [
@ -1272,33 +1293,44 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
default_namespaces.append(target_name.lower().strip())
ns_list = namespaces or default_namespaces
# Ports
default_ports = [80, 443, 8080, 8443, 3000, 4444, 5432, 6379, 9090, 9093]
port_list = ports or default_ports
# Generate all combinations grouped by namespace
# Generate URLs grouped by namespace (service-specific ports, not cartesian)
by_namespace: dict[str, list[str]] = {}
total = 0
for ns in ns_list:
urls: list[str] = []
for svc in services:
for port in port_list:
urls.append(f"http://{svc}.{ns}.svc.cluster.local:{port}")
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
# Also generate short-form names for targets that resolve short names
short_forms: list[str] = []
for svc in services:
short_forms.append(f"http://{svc}")
for svc in service_ports:
short_forms.append(f"{scheme}://{svc}")
for ns in ns_list:
short_forms.append(f"http://{svc}.{ns}")
short_forms.append(f"{scheme}://{svc}.{ns}")
return json.dumps({
# Cap output
omitted = 0
if total > max_urls:
for ns in by_namespace:
if total <= max_urls:
break
excess = total - max_urls
if excess >= len(by_namespace[ns]):
total -= len(by_namespace[ns])
omitted += len(by_namespace[ns])
by_namespace[ns] = []
else:
by_namespace[ns] = by_namespace[ns][:-excess]
omitted += excess
total -= excess
result: dict[str, Any] = {
"total_urls": total,
"services": services,
"services": list(service_ports.keys()),
"namespaces": ns_list,
"ports": port_list,
"urls_by_namespace": by_namespace,
"short_forms": short_forms,
"usage_hint": (
@ -1306,7 +1338,12 @@ def register_analysis_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
"baseline (known-bad hostname) to identify which services resolve. "
"Short forms work when K8s DNS search domains are configured."
),
})
}
if omitted:
result["omitted_urls"] = omitted
result["note"] = f"{omitted} URLs omitted due to max_urls={max_urls} cap."
return json.dumps(result)
# --- Blind SSRF Oracle Builder ---

8
strix-mcp/src/strix_mcp/tools_helpers.py
View file

@ -182,8 +182,12 @@ def build_nuclei_command(
def extract_script_urls(html: str, base_url: str) -> list[str]:
"""Extract absolute URLs of <script src="..."> tags from HTML."""
pattern = r'<script[^>]+src=["\']([^"\']+)["\']'
"""Extract absolute URLs of <script src="..."> tags from HTML.
Handles attributes like type="module" and valueless attributes
(e.g. ``crossorigin``) that appear before the ``src``.
"""
pattern = r'<script[^>]*\s+src=["\']([^"\']+)["\']'
matches = re.findall(pattern, html, re.IGNORECASE)
return [urljoin(base_url, m) for m in matches]

2
strix-mcp/src/strix_mcp/tools_recon.py
View file

@ -190,7 +190,7 @@ def register_recon_tools(mcp: FastMCP, sandbox: SandboxManager) -> None:
# Build Python script that runs inside sandbox.
# Regex patterns injected via repr() to avoid escaping issues in nested strings.
script_regex = r'<script[^>]+src=["' + "'" + r'](.[^"' + "'" + r']+)["' + "'" + r']'
script_regex = r'<script[^>]*\s+src=["\']([^"\']+)["\']'
sm_regex = r'//[#@]\s*sourceMappingURL=(\S+)'
script = (
'import json, re, sys\n'

46
strix-mcp/tests/test_tools.py
View file

@ -515,6 +515,52 @@ class TestLoadSkillTool:
assert "nuclei" in result["loaded_skills"]
assert len(result["skill_content"]["nuclei"]) > 0
@pytest.mark.asyncio
async def test_max_content_length_truncates(self, mcp_no_scan):
"""When total content exceeds max_content_length, largest skills are truncated."""
result = json.loads(_tool_text(await mcp_no_scan.call_tool("load_skill", {
"skills": "idor,xss,sql_injection",
"max_content_length": 1000,
})))
assert result["success"] is True
assert len(result["loaded_skills"]) == 3
# Total content should not exceed max_content_length (with some tolerance for min 500)
total = sum(len(c) for c in result["skill_content"].values())
# All three skills should be present in skill_content
assert "idor" in result["skill_content"]
assert "xss" in result["skill_content"]
assert "sql_injection" in result["skill_content"]
# At least one skill should have been truncated
assert "truncation_notes" in result
assert len(result["truncation_notes"]) > 0
@pytest.mark.asyncio
async def test_summary_only_mode(self, mcp_no_scan):
"""summary_only=True should return skill names without full content."""
result = json.loads(_tool_text(await mcp_no_scan.call_tool("load_skill", {
"skills": "idor,xss",
"summary_only": True,
})))
assert result["success"] is True
assert "skill_content" not in result
assert "skill_summaries" in result
assert "idor" in result["skill_summaries"]
assert "xss" in result["skill_summaries"]
# Summaries should be short strings (first line)
for summary in result["skill_summaries"].values():
assert len(summary) <= 200
@pytest.mark.asyncio
async def test_max_content_length_no_truncation_when_under(self, mcp_no_scan):
"""When content is under max_content_length, no truncation occurs."""
result = json.loads(_tool_text(await mcp_no_scan.call_tool("load_skill", {
"skills": "idor",
"max_content_length": 500000,
})))
assert result["success"] is True
assert "truncation_notes" not in result
assert "idor" in result["skill_content"]
class TestScanStateLoadedSkills:
"""Tests for the loaded_skills field on ScanState."""

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

@ -1267,10 +1267,43 @@ 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"] > 100
assert result["total_urls"] > 50
assert result["total_urls"] < 500 # no longer a cartesian product
assert "urls_by_namespace" in result
assert "kube-system" in result["urls_by_namespace"]
@pytest.mark.asyncio
async def test_uses_https_scheme_by_default(self, mcp_k8s):
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {})))
all_urls = []
for ns_urls in result["urls_by_namespace"].values():
all_urls.extend(ns_urls)
assert all(u.startswith("https://") for u in all_urls)
assert all(u.startswith("https://") for u in result["short_forms"])
@pytest.mark.asyncio
async def test_custom_scheme(self, mcp_k8s):
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"scheme": "http",
})))
all_urls = []
for ns_urls in result["urls_by_namespace"].values():
all_urls.extend(ns_urls)
assert all(u.startswith("http://") for u in all_urls)
@pytest.mark.asyncio
async def test_service_specific_ports(self, mcp_k8s):
"""Services should use their known default ports, not a cartesian product."""
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"namespaces": ["default"],
})))
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]
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
@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", {
@ -1283,16 +1316,26 @@ class TestK8sEnumerate:
assert "neon" in result["urls_by_namespace"] # namespace added
@pytest.mark.asyncio
async def test_custom_namespaces_and_ports(self, mcp_k8s):
async def test_additional_ports_appended(self, mcp_k8s):
"""User-supplied ports should be added to service defaults, not replace them."""
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"namespaces": ["custom-ns"],
"namespaces": ["default"],
"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)
urls = result["urls_by_namespace"]["default"]
# 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
assert any("grafana" in u and ":3000" in u for u in urls)
@pytest.mark.asyncio
async def test_max_urls_cap(self, mcp_k8s):
"""Output should be capped at max_urls."""
result = json.loads(_tool_text(await mcp_k8s.call_tool("k8s_enumerate", {
"max_urls": 10,
})))
total_actual = sum(len(v) for v in result["urls_by_namespace"].values())
assert total_actual <= 10
@pytest.mark.asyncio
async def test_result_structure(self, mcp_k8s):

9
strix-mcp/tests/test_tools_helpers.py
View file

@ -200,6 +200,15 @@ class TestSourcemapHelpers:
assert "https://example.com/assets/vendor.js" in urls
assert len(urls) == 3
def test_extract_script_urls_module_crossorigin(self):
"""Scripts with type='module' and valueless crossorigin should be matched."""
from strix_mcp.tools_helpers import extract_script_urls
html = '<html><script type="module" crossorigin src="/v5/assets/index-DVrLtZxj.js"></script></html>'
urls = extract_script_urls(html, "https://example.com")
assert "https://example.com/v5/assets/index-DVrLtZxj.js" in urls
assert len(urls) == 1
def test_extract_script_urls_empty(self):
"""No script tags should return empty list."""
from strix_mcp.tools_helpers import extract_script_urls