scan_mcp_url

Instructions

Implementation Reference

• server.py:395-440 (handler)

  The `scan_mcp_url` MCP tool handler function. Fetches a remote MCP server's /tools/list, scans tool names/descriptions/schemas against injection rules, and returns a structured audit report with severity counts, score (0-100), verdict, and remediation hints. Uses @mcp.tool() decorator for registration.

  @mcp.tool()
  def scan_mcp_url(url: str, api_key: str = "") -> dict:
      """
      Fetch a remote MCP server's /tools/list (or any JSON tool listing) and scan
      every tool's name + description + inputSchema against the canonical 30+
      injection-pattern rules.
  
      Returns a structured report: per-tool findings, severity counts, score
      0-100, verdict, and remediation hints. Free tier: 5 scans/day per key.
      """
      ok, msg, tier = check_access(api_key)
      if not ok:
          return {"error": msg, "upgrade": STRIPE_79}
      quota_ok, quota_msg = _consume_quota(tier, key=api_key or "anonymous")
      if not quota_ok:
          return {"error": quota_msg, "upgrade_pro": STRIPE_29, "upgrade_enterprise": STRIPE_1499}
  
      fetched, payload, fetch_msg = _http_get_json(url)
      if not fetched:
          return {"error": f"could not fetch {url}: {fetch_msg}", "tier": tier, "quota": quota_msg}
      # Try common shapes: top-level list, {"tools": [...]}, or {"result": {"tools": [...]}}
      tools = None
      if isinstance(payload, list):
          tools = payload
      elif isinstance(payload, dict):
          tools = payload.get("tools") or (payload.get("result") or {}).get("tools")
      if tools is None:
          return {"error": "no 'tools' array found in response", "tier": tier, "quota": quota_msg, "raw_keys": list(payload.keys()) if isinstance(payload, dict) else None}
  
      audit = _scan_tools_list(tools)
      score = _score(audit["severity_counts"])
      verdict = _verdict(score, audit["severity_counts"])
      return {
          "url": url,
          "scanned_at": datetime.now(timezone.utc).isoformat(),
          "tier": tier,
          "quota": quota_msg,
          "score_0_100": score,
          "verdict": verdict,
          "severity_counts": audit["severity_counts"],
          "tools_scanned": audit["tools_scanned"],
          "total_findings": audit["total_findings"],
          "findings": audit["all_findings"][:50],  # cap response size
          "rules_applied": len(INJECTION_RULES),
          "next_step": "Call signed_safety_report() with these findings to issue a procurement-grade signed cert (Pro tier)." if tier in ("pro", "enterprise") else f"Upgrade Pro £29/mo for signed safety reports: {STRIPE_29}",
      }

• server.py:392-395 (registration)

  MCP FastMCP server instantiation and the @mcp.tool() decorator that registers scan_mcp_url as an MCP tool on the 'meok-mcp-injection-scan' server.

  mcp = FastMCP("meok-mcp-injection-scan")
  
  
  @mcp.tool()

• server.py:239-268 (helper)

  _scan_text() — Core helper that runs all INJECTION_RULES against a text string (tool name, description, or schema) and returns finding dicts. Supports special handler for 'long description' rule.

  def _scan_text(text: str) -> list[dict]:
      """Run every rule against a chunk of text. Returns list of finding dicts."""
      findings: list[dict] = []
      if not text:
          return findings
      for rule in INJECTION_RULES:
          pat = rule["pattern"]
          if pat == "_long_description":
              if len(text) > 1024:
                  findings.append({
                      "rule_id": rule["id"],
                      "severity": rule["severity"],
                      "category": rule["category"],
                      "name": rule["name"],
                      "evidence": f"length={len(text)} chars",
                      "remediation": rule["remediation"],
                  })
              continue
          for m in pat.finditer(text):
              findings.append({
                  "rule_id": rule["id"],
                  "severity": rule["severity"],
                  "category": rule["category"],
                  "name": rule["name"],
                  "evidence": (m.group(0)[:80] + "…") if len(m.group(0)) > 80 else m.group(0),
                  "remediation": rule["remediation"],
              })
              # one match per rule per text is enough for the report
              break
      return findings

• server.py:271-296 (helper)

  _http_get_json() — HTTP(S) GET helper that fetches JSON from a URL with built-in SSRF protection (blocks internal/loopback IPs). Used by scan_mcp_url to fetch the remote MCP server's tools list.

  def _http_get_json(url: str, timeout: int = 8) -> tuple[bool, Any, str]:
      """Fetch JSON from a URL. SSRF-safe — blocks internal IPs."""
      try:
          parsed = urllib.parse.urlparse(url)
      except Exception as e:
          return False, None, f"invalid URL: {e}"
      host = (parsed.hostname or "").lower()
      if not host:
          return False, None, "URL missing hostname"
      if host in ("localhost", "0.0.0.0") or host.startswith(("127.", "10.", "169.254.", "192.168.")) \
              or any(host.startswith(f"172.{n}.") for n in range(16, 32)):
          return False, None, f"refused: internal/loopback host {host!r}"
      if parsed.scheme not in ("http", "https"):
          return False, None, f"refused: scheme {parsed.scheme!r} (only http/https)"
      req = urllib.request.Request(
          url, method="GET",
          headers={"User-Agent": "meok-mcp-injection-scan/1.0", "Accept": "application/json"},
      )
      try:
          with urllib.request.urlopen(req, timeout=timeout) as r:
              raw = r.read().decode("utf-8", errors="replace")
          return True, json.loads(raw), "OK"
      except json.JSONDecodeError as e:
          return False, None, f"non-JSON response: {e}"
      except Exception as e:
          return False, None, f"{type(e).__name__}: {e}"

• server.py:299-326 (helper)

  _scan_tools_list() — Iterates over a list of MCP tool definitions, scanning each tool's name, description, and inputSchema against injection rules via _scan_text(). Returns per-tool findings, severity counts, etc.

  def _scan_tools_list(tools: list) -> dict:
      """Run rules against every tool in a tools/list MCP response."""
      all_findings = []
      per_tool = {}
      if not isinstance(tools, list):
          return {"error": f"tools must be a list, got {type(tools).__name__}", "findings": []}
      for t in tools:
          if not isinstance(t, dict):
              continue
          name = t.get("name") or "<unnamed>"
          desc = t.get("description") or ""
          schema_text = json.dumps(t.get("inputSchema") or {}, separators=(",", ":"))
          tool_findings = []
          tool_findings.extend(_scan_text(name))
          tool_findings.extend(_scan_text(desc))
          tool_findings.extend(_scan_text(schema_text))
          for f in tool_findings:
              f["tool"] = name
          per_tool[name] = tool_findings
          all_findings.extend(tool_findings)
      sev_counts = Counter(f["severity"] for f in all_findings)
      return {
          "tools_scanned": len(per_tool),
          "total_findings": len(all_findings),
          "severity_counts": dict(sev_counts),
          "per_tool": per_tool,
          "all_findings": all_findings,
      }