T09 · Insecure Skill Coding Practices
Error
- Location
- scripts/ai_hive_mcp.py:20
- Finding
- Unrestricted MCP Endpoint Can Expose Credentials and Submitted Data<![CDATA[ ## Vulnerability Details **File Location**: `scripts/ai_hive_mcp.py`, lines 20–68 **Vulnerability Type**: Unvalidated network destination for authenticated requests **Risk Level**: High ### Vulnerable Code ```python MCP_URL = os.environ.get("AI_HIVE_MCP_URL", "https://ai-hive.iclip.cn/api/mcp") ORIGIN = "https://ai-hive.iclip.cn" PROTECTED_RESOURCE = f"{ORIGIN}/.well-known/oauth-protected-resource/api/mcp" AUTHORIZATION_SERVER = f"{ORIGIN}/.well-known/oauth-authorization-server" READ_ONLY_TOOLS = {"ai_hive_list_models", "ai_hive_get_task"} def fetch_json(url: str) -> dict: request = urllib.request.Request(url, headers={"accept": "application/json"}) with urllib.request.urlopen(request, timeout=20) as response: return json.loads(response.read().decode("utf-8")) def parse_payload(raw: bytes, content_type: str) -> dict: text = raw.decode("utf-8", errors="replace").strip() if "text/event-stream" in content_type or text.startswith("event:") or text.startswith("data:"): for line in text.splitlines(): if line.startswith("data:"): candidate = line[5:].strip() if candidate and candidate != "[DONE]": return json.loads(candidate) raise RuntimeError("MCP returned SSE without a parseable data event.") if not text: return {} return json.loads(text) def auth_headers() -> dict[str, str]: key = os.environ.get("AI_HIVE_API_KEY", "").strip() token = os.environ.get("AI_HIVE_ACCESS_TOKEN", "").strip() if token: return {"authorization": f"Bearer {token}"} if key: return {"x-ai-hive-api-key": key} raise SystemExit( "Missing credentials. OAuth users should complete login in their MCP client; " "tool calls from this script require AI_HIVE_API_KEY, or run doctor only." ) def post(payload: dict, session_id: str | None = None) -> tuple[dict, str | None]: headers = { "content-type": "a ...[truncated 2988 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. **Remove arbitrary endpoint overrides unless operationally required.** Use the fixed documented MCP endpoint: ```python MCP_URL = "https://ai-hive.iclip.cn/api/mcp" ``` 2. **If configurability is required, validate the destination before constructing authenticated requests.** Require: - The `https` scheme. - The exact approved hostname `ai-hive.iclip.cn`. - The expected `/api/mcp` path. - No embedded username or password. - No unexpected port. - No IP-literal or look-alike hostname. 3. **Enforce an explicit origin allowlist**, for example: ```python from urllib.parse import urlparse ALLOWED_HOSTS = {"ai-hive.iclip.cn"} def validate_mcp_url(url: str) -> str: parsed = urlparse(url) if parsed.scheme != "https": raise SystemExit("The MCP endpoint must use HTTPS.") if parsed.hostname not in ALLOWED_HOSTS: raise SystemExit("The MCP endpoint host is not approved.") if parsed.username or parsed.password: raise SystemExit("Credentials must not be embedded in the MCP URL.") if parsed.port not in (None, 443): raise SystemExit("Unexpected MCP endpoint port.") if parsed.path != "/api/mcp": raise SystemExit("Unexpected MCP endpoint path.") return url ``` 4. **Disable automatic cross-origin redirects for authenticated requests**, or implement a redirect handler that rejects any redirect whose origin differs from the validated endpoint. Never forward API keys or bearer tokens to a different origin. 5. **Separate credential attachment from generic request handling.** Only add authorization headers after the destination has passed validation. 6. **Use least-privilege, short-lived credentials where supported.** Apply narrow scopes, account-level spending limits, expiration, and prompt revocation procedures. 7. **Add automated security tests** confirming that authenticated requests reject: - ...[truncated 173 chars]
