This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new e080b06 fix: constrain Doris monitoring HTTP endpoints (#129)
e080b06 is described below
commit e080b0631b248dff4e97be2056a3aa92abf895f8
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 01:15:23 2026 +0800
fix: constrain Doris monitoring HTTP endpoints (#129)
---
.env.example | 11 +-
README.md | 25 +-
doris_mcp_server/utils/analysis_tools.py | 392 ++++++++++++------------
doris_mcp_server/utils/config.py | 58 +++-
doris_mcp_server/utils/doris_http_client.py | 394 +++++++++++++++++++++++++
doris_mcp_server/utils/monitoring_tools.py | 185 +++++++-----
test/integration/test_real_doris_transports.py | 79 ++++-
test/protocol/stdio_capability_server.py | 45 +++
test/protocol/test_mcp_v2_protocol.py | 70 +++++
test/security/test_doris_http_ssrf.py | 350 ++++++++++++++++++++++
10 files changed, 1307 insertions(+), 302 deletions(-)
diff --git a/.env.example b/.env.example
index 61e2508..6c0b1de 100644
--- a/.env.example
+++ b/.env.example
@@ -33,11 +33,18 @@ DORIS_DATABASE=information_schema
# Doris FE HTTP API port (for Profile and other HTTP APIs)
DORIS_FE_HTTP_PORT=8030
-# Doris BE (Backend) nodes configuration (optional, for external access)
-# Format: host1,host2,host3 (if empty, will use "show backends" to get BE
nodes)
+# Doris BE (Backend) HTTP allowlist for monitoring metrics.
+# Format: host1,host2,host3. If empty, BE HTTP metrics are disabled.
DORIS_BE_HOSTS=
DORIS_BE_WEBSERVER_PORT=8040
+# FE/BE HTTP safety limits. Redirects are always disabled and resolved
+# metadata/link-local addresses are always rejected.
+DORIS_HTTP_CONNECT_TIMEOUT_SECONDS=3
+DORIS_HTTP_READ_TIMEOUT_SECONDS=15
+DORIS_HTTP_TOTAL_TIMEOUT_SECONDS=30
+DORIS_HTTP_MAX_RESPONSE_BYTES=4194304
+
# Connection pool configuration
DORIS_MAX_CONNECTIONS=20
DORIS_CONNECTION_TIMEOUT=30
diff --git a/README.md b/README.md
index d06e8a0..a57c57c 100644
--- a/README.md
+++ b/README.md
@@ -264,8 +264,13 @@ cp .env.example .env
* `DORIS_DATABASE`: Default database name (default: information_schema)
* `DORIS_MIN_CONNECTIONS`: Minimum connection pool size (default: 5)
* `DORIS_MAX_CONNECTIONS`: Maximum connection pool size (default: 20)
- * `DORIS_BE_HOSTS`: BE nodes for monitoring (comma-separated, optional -
auto-discovery via SHOW BACKENDS if empty)
+ * `DORIS_FE_HTTP_PORT`: FE HTTP API port for profile, table-size, and
monitoring tools (default: 8030)
+ * `DORIS_BE_HOSTS`: Explicit BE HTTP allowlist for monitoring
(comma-separated; BE HTTP metrics are disabled when empty)
* `DORIS_BE_WEBSERVER_PORT`: BE webserver port for monitoring tools
(default: 8040)
+ * `DORIS_HTTP_CONNECT_TIMEOUT_SECONDS`: FE/BE HTTP connection timeout
(default: 3)
+ * `DORIS_HTTP_READ_TIMEOUT_SECONDS`: FE/BE HTTP socket read timeout
(default: 15)
+ * `DORIS_HTTP_TOTAL_TIMEOUT_SECONDS`: FE/BE HTTP total timeout (default:
30; hard maximum: 60)
+ * `DORIS_HTTP_MAX_RESPONSE_BYTES`: FE/BE HTTP response limit (default: 4
MiB; hard maximum: 16 MiB)
* `FE_ARROW_FLIGHT_SQL_PORT`: Frontend Arrow Flight SQL port for ADBC
(New in v0.5.0)
* `BE_ARROW_FLIGHT_SQL_PORT`: Backend Arrow Flight SQL port for ADBC
(New in v0.5.0)
* **Authentication Configuration (Enhanced in v0.6.0)**:
@@ -1475,21 +1480,21 @@ If you have further requirements for the returned
results, you can describe the
### Q: How to configure BE nodes for monitoring tools?
-**A:** Choose the appropriate configuration based on your deployment scenario:
+**A:** Configure every BE HTTP node that the MCP server may contact:
-**External Network (Manual Configuration):**
```bash
-# Manually specify BE node addresses
+# Explicitly allow these configured BE nodes
DORIS_BE_HOSTS=10.1.1.100,10.1.1.101,10.1.1.102
DORIS_BE_WEBSERVER_PORT=8040
```
-**Internal Network (Automatic Discovery):**
-```bash
-# Leave BE_HOSTS empty for auto-discovery
-# DORIS_BE_HOSTS= # Not set or empty
-# System will use 'SHOW BACKENDS' command to get internal IPs
-```
+BE HTTP endpoints are never inferred from `SHOW BACKENDS`: SQL metadata is not
+an outbound HTTP allowlist. If `DORIS_BE_HOSTS` is empty, BE HTTP metrics are
+disabled. FE and BE requests use only configured hosts and ports, pin validated
+DNS results for each request, reject metadata/link-local destinations, disable
+redirects, and enforce connection/read/total timeouts plus a response byte
+limit. Private and loopback addresses remain available for normal internal
+Doris deployments and SSH tunnels.
### Q: How to use SQL Explain/Profile files with LLM for optimization?
diff --git a/doris_mcp_server/utils/analysis_tools.py
b/doris_mcp_server/utils/analysis_tools.py
index 070e861..90ae515 100644
--- a/doris_mcp_server/utils/analysis_tools.py
+++ b/doris_mcp_server/utils/analysis_tools.py
@@ -23,11 +23,12 @@ import time
from datetime import datetime
from typing import Any, Dict, List
import uuid
-import aiohttp
import hashlib
from pathlib import Path
+from urllib.parse import quote
from .db import DorisConnectionManager
+from .doris_http_client import DorisHTTPClient
from .logger import get_logger
from .sql_security_utils import (
SQLSecurityError,
@@ -879,72 +880,64 @@ class SQLAnalyzer:
Query ID string or None if not found
"""
try:
- # Get database config
db_config = self.connection_manager.config.database
-
- # Build HTTP API URL according to official documentation
- # Reference:
https://doris.apache.org/zh-CN/docs/admin-manual/open-api/fe-http/query-profile-action#通过-trace-id-获取-query-id
- url =
f"http://{db_config.host}:{db_config.fe_http_port}/rest/v2/manager/query/trace_id/{trace_id}";
-
- # HTTP Basic Auth
- auth = aiohttp.BasicAuth(db_config.user, db_config.password)
-
- logger.info(f"Requesting query ID from: {url}")
-
- async with aiohttp.ClientSession() as session:
- async with session.get(url, auth=auth, timeout=10) as response:
- if response.status == 200:
- # Check content type first
- content_type = response.headers.get('content-type', '')
- response_text = await response.text()
- logger.info(f"Response content type: {content_type}")
- logger.info(
- "Query ID response body length: %s",
- len(response_text),
- )
-
- # Parse JSON response (regardless of content-type)
- if response_text.strip():
- try:
- import json
- result = json.loads(response_text)
- logger.info("Query ID API returned JSON")
-
- # Parse response according to Doris API format
- if result.get("code") == 0 and
result.get("data"):
- data = result["data"]
- # Data can be either a string (query_id)
or object with query_ids
- if isinstance(data, str):
- logger.info(f"Found query ID: {data}")
- return data
- elif isinstance(data, dict) and
"query_ids" in data:
- query_ids = data["query_ids"]
- if query_ids:
- query_id = query_ids[0] # Take
the first query ID
- logger.info(f"Found query ID:
{query_id}")
- return query_id
- else:
- logger.warning("No query IDs found
in response")
- else:
- logger.error(f"API returned error:
{result}")
-
- except json.JSONDecodeError as e:
- logger.error(f"Failed to parse JSON response:
{e}")
- # Fallback: try to extract query ID using regex
- import re
- query_id_pattern =
r'[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}'
- matches = re.findall(query_id_pattern,
response_text)
- if matches:
- query_id = matches[0]
- logger.info(f"Extracted query ID from
text: {query_id}")
+ http_client = DorisHTTPClient.from_database_config(db_config)
+ response = await http_client.get(
+ role="fe",
+ host=db_config.host,
+ port=db_config.fe_http_port,
+ path=(
+ "/rest/v2/manager/query/trace_id/"
+ f"{quote(trace_id, safe='')}"
+ ),
+ headers={"Accept": "application/json"},
+ )
+ if response.status == 200:
+ content_type = response.headers.get("content-type", "")
+ response_text = response.text()
+ logger.info(f"Response content type: {content_type}")
+ logger.info(
+ "Query ID response body length: %s",
+ len(response_text),
+ )
+ if response_text.strip():
+ try:
+ import json
+
+ result = json.loads(response_text)
+ logger.info("Query ID API returned JSON")
+ if result.get("code") == 0 and result.get("data"):
+ data = result["data"]
+ if isinstance(data, str):
+ logger.info(f"Found query ID: {data}")
+ return data
+ if isinstance(data, dict) and "query_ids" in data:
+ query_ids = data["query_ids"]
+ if query_ids:
+ query_id = query_ids[0]
+ logger.info(f"Found query ID: {query_id}")
return query_id
- else:
- logger.error(f"HTTP request failed with status
{response.status}")
- response_text = await response.text()
- logger.error(
- "Query ID response body omitted (length=%s)",
- len(response_text),
+ logger.warning("No query IDs found in
response")
+ else:
+ logger.error(f"API returned error: {result}")
+ except json.JSONDecodeError as e:
+ logger.error(f"Failed to parse JSON response: {e}")
+ import re
+
+ query_id_pattern = (
+ r"[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-"
+ r"[a-f0-9]{4}-[a-f0-9]{12}"
)
+ matches = re.findall(query_id_pattern, response_text)
+ if matches:
+ query_id = matches[0]
+ logger.info(f"Extracted query ID from text:
{query_id}")
+ return query_id
+ else:
+ logger.error(
+ "Query ID HTTP request failed with status %s",
+ response.status,
+ )
return None
@@ -963,79 +956,71 @@ class SQLAnalyzer:
Profile data dict or None if failed
"""
try:
- # Get database config
db_config = self.connection_manager.config.database
-
- # Try both API endpoints according to official documentation
- urls = [
-
f"http://{db_config.host}:{db_config.fe_http_port}/rest/v2/manager/query/profile/text/{query_id}";,
-
f"http://{db_config.host}:{db_config.fe_http_port}/api/profile/text?query_id={query_id}";
+ endpoints = [
+ (
+ "/rest/v2/manager/query/profile/text/"
+ f"{quote(query_id, safe='')}",
+ None,
+ ),
+ ("/api/profile/text", {"query_id": query_id}),
]
-
- # HTTP Basic Auth
- auth = aiohttp.BasicAuth(db_config.user, db_config.password)
-
- for i, url in enumerate(urls):
- logger.info(f"Requesting profile from URL {i+1}: {url}")
-
- async with aiohttp.ClientSession() as session:
- async with session.get(url, auth=auth, timeout=60) as
response:
- if response.status == 200:
- content_type =
response.headers.get('content-type', '')
- response_text = await response.text()
- logger.info(f"Profile response content type:
{content_type}")
- logger.info(f"Profile response length:
{len(response_text)}")
-
- # Handle JSON response
- if 'application/json' in content_type:
- try:
- result = await response.json()
- logger.info("Profile API returned JSON")
-
- if result.get("code") == 0 and
result.get("data"):
- profile_text =
result["data"].get("profile", "")
- return {
- "query_id": query_id,
- "profile_text": profile_text,
- "profile_size": len(profile_text),
- "retrieved_at":
datetime.now().isoformat(),
- "api_endpoint": url
- }
- else:
- logger.warning(f"Profile API returned
error: {result}")
- continue # Try next URL
-
- except Exception as e:
- logger.error(f"Failed to parse profile
JSON: {e}")
- continue
-
- # Handle plain text response
- else:
- if response_text.strip() and "not found" not
in response_text.lower():
- return {
- "query_id": query_id,
- "profile_text": response_text,
- "profile_size": len(response_text),
- "retrieved_at":
datetime.now().isoformat(),
- "api_endpoint": url
- }
- else:
- logger.warning(
- "Profile not found or empty"
- )
- continue # Try next URL
-
- elif response.status == 404:
- logger.warning(f"Profile not found (404) at {url}")
- continue # Try next URL
- else:
- logger.error(f"Profile HTTP request failed with
status {response.status} at {url}")
- response_text = await response.text()
- logger.error(
- "Profile response body omitted (length=%s)",
- len(response_text),
- )
- continue # Try next URL
+ http_client = DorisHTTPClient.from_database_config(db_config)
+ for i, (path, params) in enumerate(endpoints):
+ logger.info("Requesting profile from configured FE endpoint
%s", i + 1)
+ response = await http_client.get(
+ role="fe",
+ host=db_config.host,
+ port=db_config.fe_http_port,
+ path=path,
+ params=params,
+ headers={"Accept": "application/json, text/plain"},
+ )
+ if response.status == 200:
+ content_type = response.headers.get("content-type", "")
+ response_text = response.text()
+ logger.info(f"Profile response content type:
{content_type}")
+ logger.info(f"Profile response length:
{len(response_text)}")
+ if "application/json" in content_type:
+ try:
+ import json
+
+ result = json.loads(response_text)
+ logger.info("Profile API returned JSON")
+ if result.get("code") == 0 and result.get("data"):
+ profile_text = result["data"].get("profile",
"")
+ return {
+ "query_id": query_id,
+ "profile_text": profile_text,
+ "profile_size": len(profile_text),
+ "retrieved_at": datetime.now().isoformat(),
+ "api_endpoint": response.url,
+ }
+ logger.warning(f"Profile API returned error:
{result}")
+ continue
+ except Exception as e:
+ logger.error(f"Failed to parse profile JSON: {e}")
+ continue
+ if (
+ response_text.strip()
+ and "not found" not in response_text.lower()
+ ):
+ return {
+ "query_id": query_id,
+ "profile_text": response_text,
+ "profile_size": len(response_text),
+ "retrieved_at": datetime.now().isoformat(),
+ "api_endpoint": response.url,
+ }
+ logger.warning("Profile not found or empty")
+ continue
+ if response.status == 404:
+ logger.warning("Profile not found (404)")
+ continue
+ logger.error(
+ "Profile HTTP request failed with status %s",
+ response.status,
+ )
return None
@@ -1061,14 +1046,7 @@ class SQLAnalyzer:
Dict containing table data size information
"""
try:
- # Get database config
db_config = self.connection_manager.config.database
-
- # Build HTTP API URL according to official documentation
- # Reference:
https://doris.apache.org/zh-CN/docs/admin-manual/open-api/fe-http/show-table-data-action
- url =
f"http://{db_config.host}:{db_config.fe_http_port}/api/show_table_data";
-
- # Build query parameters
params = {}
if db_name:
params["db"] = db_name
@@ -1076,72 +1054,76 @@ class SQLAnalyzer:
params["table"] = table_name
if single_replica:
params["single_replica"] = "true"
-
- # HTTP Basic Auth
- auth = aiohttp.BasicAuth(db_config.user, db_config.password)
-
- logger.info(f"Requesting table data size from: {url} with params:
{params}")
-
- async with aiohttp.ClientSession() as session:
- async with session.get(url, auth=auth, params=params,
timeout=30) as response:
- if response.status == 200:
- response_text = await response.text()
- logger.info(f"Table data size response length:
{len(response_text)}")
-
- try:
- # Parse JSON response
- import json
- result = json.loads(response_text)
-
- if result.get("code") == 0 and result.get("data"):
- data = result["data"]
-
- # Process and format the data
- formatted_data =
self._format_table_data_size(data, db_name, table_name, single_replica)
-
- return {
- "success": True,
- "db_name": db_name,
- "table_name": table_name,
- "single_replica": single_replica,
- "timestamp": datetime.now().isoformat(),
- "data": formatted_data,
- "url": url,
- "note": "Table data size information from
Doris FE HTTP API"
- }
- else:
- return {
- "success": False,
- "error": f"API returned error: {result}",
- "db_name": db_name,
- "table_name": table_name,
- "url": url,
- "timestamp": datetime.now().isoformat()
- }
-
- except json.JSONDecodeError as e:
- logger.error(f"Failed to parse JSON response: {e}")
- return {
- "success": False,
- "error": f"Failed to parse JSON response: {e}",
- "response_text": response_text[:500], # First
500 chars for debugging
- "url": url,
- "timestamp": datetime.now().isoformat()
- }
- else:
- logger.error(f"HTTP request failed with status
{response.status}")
- response_text = await response.text()
- logger.error(
- "Table size response body omitted (length=%s)",
- len(response_text),
+ logger.info(
+ "Requesting table data size from configured FE endpoint with
params: %s",
+ params,
+ )
+ http_client = DorisHTTPClient.from_database_config(db_config)
+ response = await http_client.get(
+ role="fe",
+ host=db_config.host,
+ port=db_config.fe_http_port,
+ path="/api/show_table_data",
+ params=params,
+ headers={"Accept": "application/json"},
+ )
+ if response.status == 200:
+ response_text = response.text()
+ logger.info(
+ "Table data size response length: %s",
+ len(response_text),
+ )
+ try:
+ import json
+
+ result = json.loads(response_text)
+ if result.get("code") == 0 and result.get("data"):
+ data = result["data"]
+ formatted_data = self._format_table_data_size(
+ data,
+ db_name,
+ table_name,
+ single_replica,
)
return {
- "success": False,
- "error": f"HTTP request failed with status
{response.status}",
- "response_text": response_text[:500], # First 500
chars for debugging
- "url": url,
- "timestamp": datetime.now().isoformat()
+ "success": True,
+ "db_name": db_name,
+ "table_name": table_name,
+ "single_replica": single_replica,
+ "timestamp": datetime.now().isoformat(),
+ "data": formatted_data,
+ "url": response.url,
+ "note": (
+ "Table data size information from Doris FE "
+ "HTTP API"
+ ),
}
+ return {
+ "success": False,
+ "error": f"API returned error: {result}",
+ "db_name": db_name,
+ "table_name": table_name,
+ "url": response.url,
+ "timestamp": datetime.now().isoformat(),
+ }
+ except json.JSONDecodeError as e:
+ logger.error(f"Failed to parse JSON response: {e}")
+ return {
+ "success": False,
+ "error": f"Failed to parse JSON response: {e}",
+ "url": response.url,
+ "timestamp": datetime.now().isoformat(),
+ }
+ logger.error(
+ "Table data size HTTP request failed with status %s",
+ response.status,
+ )
+ return {
+ "success": False,
+ "error": f"HTTP request failed with status {response.status}",
+ "url": response.url,
+ "timestamp": datetime.now().isoformat(),
+ }
except Exception as e:
logger.error(f"Table data size request failed: {str(e)}")
diff --git a/doris_mcp_server/utils/config.py b/doris_mcp_server/utils/config.py
index 001f6b5..c59e963 100644
--- a/doris_mcp_server/utils/config.py
+++ b/doris_mcp_server/utils/config.py
@@ -285,11 +285,18 @@ class DatabaseConfig:
# FE HTTP API port for profile and other HTTP APIs
fe_http_port: int = 8030
- # BE nodes configuration for external access
- # If be_hosts is empty, will use "show backends" to get BE nodes
+ # BE HTTP nodes must be configured explicitly. SQL metadata is not trusted
+ # as an outbound HTTP allowlist.
be_hosts: list[str] = field(default_factory=list)
be_webserver_port: int = 8040
+ # Shared FE/BE HTTP safety limits. Runtime enforcement also applies hard
+ # caps so unsafe config values cannot remove the boundary.
+ http_connect_timeout_seconds: float = 3.0
+ http_read_timeout_seconds: float = 15.0
+ http_total_timeout_seconds: float = 30.0
+ http_max_response_bytes: int = 4 * 1024 * 1024
+
# Arrow Flight SQL Configuration (Required for ADBC tools)
fe_arrow_flight_sql_port: int | None = None
be_arrow_flight_sql_port: int | None = None
@@ -696,6 +703,31 @@ class DorisConfig:
be_webserver_port = os.getenv("DORIS_BE_WEBSERVER_PORT", "").strip()
if be_webserver_port and be_webserver_port.isdigit():
config.database.be_webserver_port = int(be_webserver_port)
+
+ config.database.http_connect_timeout_seconds = float(
+ os.getenv(
+ "DORIS_HTTP_CONNECT_TIMEOUT_SECONDS",
+ str(config.database.http_connect_timeout_seconds),
+ )
+ )
+ config.database.http_read_timeout_seconds = float(
+ os.getenv(
+ "DORIS_HTTP_READ_TIMEOUT_SECONDS",
+ str(config.database.http_read_timeout_seconds),
+ )
+ )
+ config.database.http_total_timeout_seconds = float(
+ os.getenv(
+ "DORIS_HTTP_TOTAL_TIMEOUT_SECONDS",
+ str(config.database.http_total_timeout_seconds),
+ )
+ )
+ config.database.http_max_response_bytes = int(
+ os.getenv(
+ "DORIS_HTTP_MAX_RESPONSE_BYTES",
+ str(config.database.http_max_response_bytes),
+ )
+ )
# Arrow Flight SQL Configuration
fe_arrow_port_env = os.getenv("FE_ARROW_FLIGHT_SQL_PORT")
@@ -1181,6 +1213,10 @@ class DorisConfig:
"fe_http_port": self.database.fe_http_port,
"be_hosts": self.database.be_hosts,
"be_webserver_port": self.database.be_webserver_port,
+ "http_connect_timeout_seconds":
self.database.http_connect_timeout_seconds,
+ "http_read_timeout_seconds":
self.database.http_read_timeout_seconds,
+ "http_total_timeout_seconds":
self.database.http_total_timeout_seconds,
+ "http_max_response_bytes":
self.database.http_max_response_bytes,
"fe_arrow_flight_sql_port":
self.database.fe_arrow_flight_sql_port,
"be_arrow_flight_sql_port":
self.database.be_arrow_flight_sql_port,
"min_connections": self.database.min_connections, # Always 0,
shown for reference
@@ -1328,6 +1364,24 @@ class DorisConfig:
if self.database.max_connections <= 0:
errors.append("Maximum connections must be greater than 0")
+ if not (1 <= self.database.fe_http_port <= 65535):
+ errors.append("Doris FE HTTP port must be in the range 1-65535")
+
+ if not (1 <= self.database.be_webserver_port <= 65535):
+ errors.append("Doris BE HTTP port must be in the range 1-65535")
+
+ if self.database.http_connect_timeout_seconds <= 0:
+ errors.append("Doris HTTP connect timeout must be greater than 0")
+
+ if self.database.http_read_timeout_seconds <= 0:
+ errors.append("Doris HTTP read timeout must be greater than 0")
+
+ if self.database.http_total_timeout_seconds <= 0:
+ errors.append("Doris HTTP total timeout must be greater than 0")
+
+ if self.database.http_max_response_bytes <= 0:
+ errors.append("Doris HTTP response byte limit must be greater than
0")
+
# Validate security configuration
if self.security.auth_type not in ["token", "basic", "oauth", "jwt"]:
errors.append("Authentication type must be one of token, basic,
oauth, or jwt")
diff --git a/doris_mcp_server/utils/doris_http_client.py
b/doris_mcp_server/utils/doris_http_client.py
new file mode 100644
index 0000000..7406e9f
--- /dev/null
+++ b/doris_mcp_server/utils/doris_http_client.py
@@ -0,0 +1,394 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Bounded HTTP access to explicitly configured Doris FE and BE nodes."""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import socket
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Literal
+from urllib.parse import urlencode, urlunsplit
+
+import aiohttp
+
+DEFAULT_CONNECT_TIMEOUT_SECONDS = 3.0
+DEFAULT_READ_TIMEOUT_SECONDS = 15.0
+DEFAULT_TOTAL_TIMEOUT_SECONDS = 30.0
+DEFAULT_MAX_RESPONSE_BYTES = 4 * 1024 * 1024
+MAX_TIMEOUT_SECONDS = 60.0
+MAX_RESPONSE_BYTES = 16 * 1024 * 1024
+
+_METADATA_HOSTS = frozenset(
+ {
+ "metadata",
+ "metadata.google.internal",
+ "metadata.goog",
+ "instance-data",
+ }
+)
+_METADATA_ADDRESSES = frozenset(
+ {
+ ipaddress.ip_address("169.254.169.254"),
+ ipaddress.ip_address("169.254.170.2"),
+ ipaddress.ip_address("100.100.100.200"),
+ ipaddress.ip_address("fd00:ec2::254"),
+ }
+)
+
+
+class DorisHTTPError(RuntimeError):
+ """Base error for a Doris HTTP request."""
+
+
+class DorisHTTPPolicyError(DorisHTTPError):
+ """The requested endpoint violates the configured SSRF boundary."""
+
+
+class DorisHTTPResponseTooLarge(DorisHTTPError):
+ """The Doris HTTP response exceeded the configured byte limit."""
+
+
+class DorisHTTPRequestError(DorisHTTPError):
+ """The configured Doris HTTP endpoint could not be reached."""
+
+
+@dataclass(frozen=True)
+class DorisHTTPResponse:
+ """A bounded in-memory Doris HTTP response."""
+
+ status: int
+ headers: Mapping[str, str]
+ body: bytes
+ url: str
+
+ def text(self) -> str:
+ return self.body.decode("utf-8", errors="replace")
+
+
+class _PinnedResolver(aiohttp.abc.AbstractResolver):
+ """Resolve one configured hostname only to preflighted addresses."""
+
+ def __init__(self, hostname: str, addresses: tuple[str, ...]):
+ self.hostname = hostname
+ self.addresses = addresses
+
+ async def resolve(
+ self,
+ host: str,
+ port: int = 0,
+ family: socket.AddressFamily = socket.AF_INET,
+ ) -> list[aiohttp.abc.ResolveResult]:
+ if _normalize_host(host) != self.hostname:
+ raise OSError("Unexpected Doris HTTP hostname")
+ results: list[aiohttp.abc.ResolveResult] = []
+ for address in self.addresses:
+ parsed = ipaddress.ip_address(address)
+ address_family = socket.AF_INET6 if parsed.version == 6 else
socket.AF_INET
+ if family not in {socket.AF_UNSPEC, address_family}:
+ continue
+ results.append(
+ {
+ "hostname": host,
+ "host": address,
+ "port": port,
+ "family": address_family,
+ "proto": socket.IPPROTO_TCP,
+ "flags": socket.AI_NUMERICHOST,
+ }
+ )
+ if not results:
+ raise OSError("No validated Doris HTTP address")
+ return results
+
+ async def close(self) -> None:
+ return None
+
+
+def _normalize_host(host: Any) -> str:
+ if not isinstance(host, str):
+ raise DorisHTTPPolicyError("Doris HTTP host must be a string")
+ normalized = host.strip()
+ if normalized.startswith("[") and normalized.endswith("]"):
+ normalized = normalized[1:-1]
+ normalized = normalized.rstrip(".")
+ if (
+ not normalized
+ or "://" in normalized
+ or any(character in normalized for character in "/\\?#@%")
+ ):
+ raise DorisHTTPPolicyError("Doris HTTP host is malformed")
+ try:
+ return normalized.encode("idna").decode("ascii").lower()
+ except UnicodeError as exc:
+ raise DorisHTTPPolicyError("Doris HTTP host is malformed") from exc
+
+
+def _validate_port(port: Any) -> int:
+ if isinstance(port, bool) or not isinstance(port, int) or not (1 <= port
<= 65535):
+ raise DorisHTTPPolicyError("Doris HTTP port must be between 1 and
65535")
+ return port
+
+
+def _address_allowed(address: str) -> bool:
+ parsed = ipaddress.ip_address(address)
+ candidates = (parsed, getattr(parsed, "ipv4_mapped", None))
+ return not any(
+ candidate is not None
+ and (
+ candidate in _METADATA_ADDRESSES
+ or candidate.is_link_local
+ or candidate.is_multicast
+ or candidate.is_unspecified
+ or candidate.is_reserved
+ )
+ for candidate in candidates
+ )
+
+
+def _bounded_float(value: Any, default: float) -> float:
+ try:
+ parsed = float(value)
+ except (TypeError, ValueError):
+ return default
+ if parsed <= 0:
+ return default
+ return min(parsed, MAX_TIMEOUT_SECONDS)
+
+
+def _bounded_response_limit(value: Any) -> int:
+ try:
+ parsed = int(value)
+ except (TypeError, ValueError):
+ return DEFAULT_MAX_RESPONSE_BYTES
+ if parsed <= 0:
+ return DEFAULT_MAX_RESPONSE_BYTES
+ return min(parsed, MAX_RESPONSE_BYTES)
+
+
+class DorisHTTPClient:
+ """Fetch only configured Doris HTTP endpoints with SSRF controls."""
+
+ def __init__(
+ self,
+ *,
+ user: str,
+ password: str,
+ allowed_endpoints: Mapping[str, set[tuple[str, int]]],
+ connect_timeout_seconds: float = DEFAULT_CONNECT_TIMEOUT_SECONDS,
+ read_timeout_seconds: float = DEFAULT_READ_TIMEOUT_SECONDS,
+ total_timeout_seconds: float = DEFAULT_TOTAL_TIMEOUT_SECONDS,
+ max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
+ ):
+ self.user = user
+ self.password = password
+ self.allowed_endpoints = {
+ role: {
+ (_normalize_host(host), _validate_port(port))
+ for host, port in endpoints
+ }
+ for role, endpoints in allowed_endpoints.items()
+ }
+ self.connect_timeout_seconds = _bounded_float(
+ connect_timeout_seconds,
+ DEFAULT_CONNECT_TIMEOUT_SECONDS,
+ )
+ self.read_timeout_seconds = _bounded_float(
+ read_timeout_seconds,
+ DEFAULT_READ_TIMEOUT_SECONDS,
+ )
+ self.total_timeout_seconds = _bounded_float(
+ total_timeout_seconds,
+ DEFAULT_TOTAL_TIMEOUT_SECONDS,
+ )
+ self.max_response_bytes = _bounded_response_limit(max_response_bytes)
+
+ @classmethod
+ def from_database_config(cls, database_config: Any) -> DorisHTTPClient:
+ fe_host = _normalize_host(database_config.host)
+ fe_port = _validate_port(database_config.fe_http_port)
+ be_port = _validate_port(getattr(database_config, "be_webserver_port",
8040))
+ be_hosts = getattr(database_config, "be_hosts", []) or []
+ if not isinstance(be_hosts, list) or any(
+ not isinstance(host, str) for host in be_hosts
+ ):
+ raise DorisHTTPPolicyError("Configured Doris BE hosts are invalid")
+ return cls(
+ user=str(database_config.user),
+ password=str(database_config.password),
+ allowed_endpoints={
+ "fe": {(fe_host, fe_port)},
+ "be": {(_normalize_host(host), be_port) for host in be_hosts},
+ },
+ connect_timeout_seconds=getattr(
+ database_config,
+ "http_connect_timeout_seconds",
+ DEFAULT_CONNECT_TIMEOUT_SECONDS,
+ ),
+ read_timeout_seconds=getattr(
+ database_config,
+ "http_read_timeout_seconds",
+ DEFAULT_READ_TIMEOUT_SECONDS,
+ ),
+ total_timeout_seconds=getattr(
+ database_config,
+ "http_total_timeout_seconds",
+ DEFAULT_TOTAL_TIMEOUT_SECONDS,
+ ),
+ max_response_bytes=getattr(
+ database_config,
+ "http_max_response_bytes",
+ DEFAULT_MAX_RESPONSE_BYTES,
+ ),
+ )
+
+ async def get(
+ self,
+ *,
+ role: Literal["fe", "be"],
+ host: str,
+ port: int,
+ path: str,
+ params: Mapping[str, str] | None = None,
+ headers: Mapping[str, str] | None = None,
+ ) -> DorisHTTPResponse:
+ normalized_host = _normalize_host(host)
+ validated_port = _validate_port(port)
+ if (normalized_host, validated_port) not in self.allowed_endpoints.get(
+ role,
+ set(),
+ ):
+ raise DorisHTTPPolicyError(
+ "Doris HTTP request is not an explicitly configured endpoint"
+ )
+ if normalized_host in _METADATA_HOSTS:
+ raise DorisHTTPPolicyError(
+ "Doris HTTP endpoint is prohibited by SSRF policy"
+ )
+ if (
+ not isinstance(path, str)
+ or not path.startswith("/")
+ or path.startswith("//")
+ or any(character in path for character in "\\?#\r\n")
+ ):
+ raise DorisHTTPPolicyError("Doris HTTP path is malformed")
+
+ addresses = await self._resolve_addresses(normalized_host,
validated_port)
+ resolver = _PinnedResolver(normalized_host, addresses)
+ connector = aiohttp.TCPConnector(
+ resolver=resolver,
+ use_dns_cache=False,
+ ttl_dns_cache=0,
+ force_close=True,
+ limit=1,
+ )
+ timeout = aiohttp.ClientTimeout(
+ total=self.total_timeout_seconds,
+ connect=self.connect_timeout_seconds,
+ sock_connect=self.connect_timeout_seconds,
+ sock_read=self.read_timeout_seconds,
+ )
+ host_for_url = (
+ f"[{normalized_host}]" if ":" in normalized_host else
normalized_host
+ )
+ query = urlencode(params or {}, doseq=False)
+ url = urlunsplit(("http", f"{host_for_url}:{validated_port}", path,
query, ""))
+ auth = aiohttp.BasicAuth(self.user, self.password)
+ try:
+ async with aiohttp.ClientSession(
+ connector=connector,
+ timeout=timeout,
+ raise_for_status=False,
+ trust_env=False,
+ ) as session:
+ async with session.get(
+ url,
+ auth=auth,
+ allow_redirects=False,
+ headers=dict(headers or {}),
+ ) as response:
+ body = await self._read_bounded_body(response)
+ return DorisHTTPResponse(
+ status=response.status,
+ headers=dict(response.headers),
+ body=body,
+ url=str(response.url),
+ )
+ except DorisHTTPError:
+ raise
+ except TimeoutError as exc:
+ raise DorisHTTPRequestError("Doris HTTP request timed out") from
exc
+ except (aiohttp.ClientError, OSError) as exc:
+ raise DorisHTTPRequestError("Doris HTTP request failed") from exc
+
+ async def _resolve_addresses(self, host: str, port: int) -> tuple[str,
...]:
+ try:
+ literal = ipaddress.ip_address(host)
+ except ValueError:
+ literal = None
+ if literal is not None:
+ addresses = (str(literal),)
+ else:
+ loop = asyncio.get_running_loop()
+ try:
+ results = await asyncio.wait_for(
+ loop.getaddrinfo(
+ host,
+ port,
+ family=socket.AF_UNSPEC,
+ type=socket.SOCK_STREAM,
+ ),
+ timeout=self.connect_timeout_seconds,
+ )
+ except (OSError, TimeoutError) as exc:
+ raise DorisHTTPRequestError(
+ "Doris HTTP hostname resolution failed"
+ ) from exc
+ addresses = tuple(
+ dict.fromkeys(result[4][0].split("%", 1)[0] for result in
results)
+ )
+ if not addresses or any(not _address_allowed(address) for address in
addresses):
+ raise DorisHTTPPolicyError(
+ "Doris HTTP endpoint resolves to a prohibited address"
+ )
+ return addresses
+
+ async def _read_bounded_body(
+ self,
+ response: aiohttp.ClientResponse,
+ ) -> bytes:
+ content_length = response.headers.get("Content-Length")
+ if content_length:
+ try:
+ if int(content_length) > self.max_response_bytes:
+ raise DorisHTTPResponseTooLarge(
+ "Doris HTTP response exceeds the configured byte limit"
+ )
+ except ValueError as exc:
+ raise DorisHTTPRequestError(
+ "Doris HTTP response has an invalid Content-Length"
+ ) from exc
+ body = bytearray()
+ async for chunk in response.content.iter_chunked(64 * 1024):
+ body.extend(chunk)
+ if len(body) > self.max_response_bytes:
+ raise DorisHTTPResponseTooLarge(
+ "Doris HTTP response exceeds the configured byte limit"
+ )
+ return bytes(body)
diff --git a/doris_mcp_server/utils/monitoring_tools.py
b/doris_mcp_server/utils/monitoring_tools.py
index d9055c2..ba32b94 100644
--- a/doris_mcp_server/utils/monitoring_tools.py
+++ b/doris_mcp_server/utils/monitoring_tools.py
@@ -20,15 +20,18 @@ Provides monitoring and metrics collection functions for FE
and BE nodes
"""
import re
-import aiohttp
-import asyncio
from enum import Enum
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Literal, Optional, Tuple
from datetime import datetime
from .db import DorisConnectionManager
+from .doris_http_client import (
+ DorisHTTPClient,
+ DorisHTTPPolicyError,
+ DorisHTTPRequestError,
+ DorisHTTPResponseTooLarge,
+)
from .logger import get_logger
-from .sql_security_utils import get_auth_context
logger = get_logger(__name__)
@@ -683,16 +686,14 @@ class DorisMonitoringTools:
self.connection_manager = connection_manager
async def get_be_nodes(self) -> List[Dict[str, Any]]:
- """Get BE node information, prioritize configured be_hosts, fallback
to SHOW BACKENDS"""
+ """Return only BE HTTP nodes from the explicit configuration
allowlist."""
try:
- # Get database configuration
db_config = self.connection_manager.config.database
-
- # Check if BE hosts are configured
- if db_config.be_hosts:
- logger.info(f"Using configured BE hosts: {db_config.be_hosts}")
+ be_hosts = getattr(db_config, "be_hosts", []) or []
+ if be_hosts:
+ logger.info(f"Using configured BE hosts: {be_hosts}")
be_nodes = []
- for i, host in enumerate(db_config.be_hosts):
+ for i, host in enumerate(be_hosts):
be_info = {
"backend_id": f"configured_{i}",
"host": host,
@@ -710,80 +711,91 @@ class DorisMonitoringTools:
logger.info(f"Found {len(be_nodes)} configured BE nodes")
return be_nodes
-
- # Fallback to SHOW BACKENDS if no BE hosts configured
- logger.info("No BE hosts configured, using SHOW BACKENDS to
discover BE nodes")
- connection = await self.connection_manager.get_connection("query")
- auth_context = get_auth_context()
- result = await connection.execute("SHOW BACKENDS",
auth_context=auth_context)
-
- be_nodes = []
- for row in result.data:
- # SHOW BACKENDS returns columns including: BackendId, Host,
HeartbeatPort, BePort, HttpPort, BrpcPort, etc.
- be_info = {
- "backend_id": row.get("BackendId"),
- "host": row.get("Host"),
- "heartbeat_port": row.get("HeartbeatPort"),
- "be_port": row.get("BePort"),
- "http_port": row.get("HttpPort"), # This is webserver_port
- "brpc_port": row.get("BrpcPort"),
- "alive": row.get("Alive"),
- "system_decommissioned": row.get("SystemDecommissioned"),
- "cluster_id": row.get("ClusterId"),
- "version": row.get("Version"),
- "source": "show_backends"
- }
- be_nodes.append(be_info)
-
- logger.info(f"Found {len(be_nodes)} BE nodes from SHOW BACKENDS")
- return be_nodes
+ logger.warning(
+ "BE HTTP metrics disabled because DORIS_BE_HOSTS is empty"
+ )
+ return []
except Exception as e:
logger.error(f"Failed to get BE nodes: {str(e)}")
return []
- async def fetch_metrics_from_url(self, url: str, node_type: str,
node_info: Dict[str, Any]) -> Dict[str, Any]:
- """Fetch monitoring metrics from specified URL"""
+ async def fetch_metrics_from_node(
+ self,
+ node_type: Literal["fe", "be"],
+ node_info: Dict[str, Any],
+ ) -> Dict[str, Any]:
+ """Fetch metrics from an explicitly configured FE or BE node."""
+ endpoint = f"{node_info.get('host')}:{node_info.get('port') or
node_info.get('http_port')}"
try:
- # Get database configuration for authentication
db_config = self.connection_manager.config.database
- auth = aiohttp.BasicAuth(db_config.user, db_config.password)
-
- logger.info(f"Fetching metrics from {node_type} node: {url}")
-
- async with aiohttp.ClientSession() as session:
- async with session.get(url, auth=auth, timeout=30) as response:
- if response.status == 200:
- # Parse Prometheus format
- metrics_text = await response.text()
- metrics_data =
self._parse_prometheus_metrics(metrics_text)
-
- return {
- "success": True,
- "node_type": node_type,
- "node_info": node_info,
- "metrics": metrics_data,
- "url": url,
- "timestamp": datetime.now().isoformat()
- }
- else:
- logger.error(f"HTTP request failed with status
{response.status} for {url}")
- return {
- "success": False,
- "error": f"HTTP {response.status}",
- "node_type": node_type,
- "node_info": node_info,
- "url": url
- }
-
+ port_key = "port" if node_type == "fe" else "http_port"
+ http_client = DorisHTTPClient.from_database_config(db_config)
+ response = await http_client.get(
+ role=node_type,
+ host=node_info["host"],
+ port=int(node_info[port_key]),
+ path="/metrics",
+ headers={"Accept": "text/plain"},
+ )
+ if response.status == 200:
+ metrics_data = self._parse_prometheus_metrics(response.text())
+ return {
+ "success": True,
+ "node_type": node_type,
+ "node_info": node_info,
+ "metrics": metrics_data,
+ "url": response.url,
+ "timestamp": datetime.now().isoformat(),
+ }
+ logger.error(
+ "HTTP metrics request failed with status %s for %s",
+ response.status,
+ endpoint,
+ )
+ return {
+ "success": False,
+ "error": f"HTTP {response.status}",
+ "error_type": "http_status",
+ "node_type": node_type,
+ "node_info": node_info,
+ "url": response.url,
+ }
+ except DorisHTTPPolicyError as e:
+ logger.error("Blocked Doris metrics endpoint %s: %s", endpoint, e)
+ return {
+ "success": False,
+ "error": str(e),
+ "error_type": "prohibited_endpoint",
+ "node_type": node_type,
+ "node_info": node_info,
+ }
+ except DorisHTTPResponseTooLarge as e:
+ logger.error("Doris metrics response too large for %s", endpoint)
+ return {
+ "success": False,
+ "error": str(e),
+ "error_type": "response_too_large",
+ "node_type": node_type,
+ "node_info": node_info,
+ }
+ except DorisHTTPRequestError as e:
+ logger.error("Doris metrics request failed for %s: %s", endpoint,
e)
+ return {
+ "success": False,
+ "error": str(e),
+ "error_type": "request_failed",
+ "node_type": node_type,
+ "node_info": node_info,
+ }
except Exception as e:
- logger.error(f"Failed to fetch metrics from {url}: {str(e)}")
+ logger.error("Failed to fetch metrics from %s: %s", endpoint, e)
return {
"success": False,
"error": str(e),
+ "error_type": "internal_error",
"node_type": node_type,
"node_info": node_info,
- "url": url
}
def _parse_prometheus_metrics(self, metrics_text: str) -> Dict[str, Any]:
@@ -955,13 +967,12 @@ class DorisMonitoringTools:
"""Get FE monitoring metrics"""
try:
db_config = self.connection_manager.config.database
- fe_url =
f"http://{db_config.host}:{db_config.fe_http_port}/metrics";
-
- fe_result = await self.fetch_metrics_from_url(
- fe_url,
- "fe",
+ fe_result = await self.fetch_metrics_from_node(
+ "fe",
{"host": db_config.host, "port": db_config.fe_http_port}
)
+ if not fe_result.get("success"):
+ return fe_result
if fe_result.get("success") and priority == "p0":
fe_p0_metrics = self._get_metrics_by_type("fe", monitor_type)
@@ -999,13 +1010,25 @@ class DorisMonitoringTools:
"""Get BE monitoring metrics from all BE nodes"""
try:
be_nodes = await self.get_be_nodes()
+ if not be_nodes:
+ return [
+ {
+ "success": False,
+ "error": (
+ "BE HTTP metrics require explicit DORIS_BE_HOSTS "
+ "configuration"
+ ),
+ "error_type": "unconfigured_endpoint",
+ }
+ ]
be_results = []
for be_node in be_nodes:
if be_node.get("alive") == "true": # Only get alive BE nodes
- be_url =
f"http://{be_node['host']}:{be_node['http_port']}/metrics"
-
- be_result = await self.fetch_metrics_from_url(be_url,
"be", be_node)
+ be_result = await self.fetch_metrics_from_node("be",
be_node)
+ if not be_result.get("success"):
+ be_results.append(be_result)
+ continue
if be_result.get("success") and priority == "p0":
be_p0_metrics = self._get_metrics_by_type("be",
monitor_type)
@@ -1608,4 +1631,4 @@ class DorisMonitoringTools:
return total_bytes
except (ValueError, TypeError):
- return None
\ No newline at end of file
+ return None
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index 2f703fd..0e1f128 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -19,7 +19,10 @@
Set ``DORIS_REAL_INTEGRATION=1`` and provide ``DORIS_REAL_HOST``,
``DORIS_REAL_USER``, and ``DORIS_REAL_DATABASE``. ``DORIS_REAL_PORT`` defaults
-to 9030 and ``DORIS_REAL_PASSWORD`` may be empty.
+to 9030 and ``DORIS_REAL_PASSWORD`` may be empty. Set
+``DORIS_REAL_HTTP_INTEGRATION=1`` with ``DORIS_REAL_FE_HTTP_PORT``,
+``DORIS_REAL_BE_HOSTS``, and ``DORIS_REAL_BE_WEBSERVER_PORT`` to exercise the
+configured FE/BE HTTP monitoring boundary through both transports.
"""
from __future__ import annotations
@@ -172,7 +175,7 @@ def _server_environment(
user: str,
password: str,
) -> dict[str, str]:
- return {
+ environment = {
**os.environ,
"DORIS_HOST": settings.host,
"DORIS_PORT": str(settings.port),
@@ -190,6 +193,16 @@ def _server_environment(
"LOG_LEVEL": "ERROR",
"PYTHONUNBUFFERED": "1",
}
+ fe_http_port = os.getenv("DORIS_REAL_FE_HTTP_PORT", "").strip()
+ if fe_http_port:
+ environment["DORIS_FE_HTTP_PORT"] = fe_http_port
+ be_hosts = os.getenv("DORIS_REAL_BE_HOSTS", "").strip()
+ if be_hosts:
+ environment["DORIS_BE_HOSTS"] = be_hosts
+ be_http_port = os.getenv("DORIS_REAL_BE_WEBSERVER_PORT", "").strip()
+ if be_http_port:
+ environment["DORIS_BE_WEBSERVER_PORT"] = be_http_port
+ return environment
def _free_tcp_port() -> int:
@@ -572,3 +585,65 @@ async def test_real_doris_tool_regression_paths(
assert recovered_result.is_error is False
assert recovered_payload["success"] is True
assert recovered_payload["data"][0]["recovered"] == 1
+
+
[email protected](
+ os.getenv("DORIS_REAL_HTTP_INTEGRATION") != "1",
+ reason="set DORIS_REAL_HTTP_INTEGRATION=1 with FE/BE HTTP endpoints",
+)
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_configured_monitoring_http_endpoints_and_recovery(
+ transport: str,
+ doris_sandbox: DorisSandbox,
+) -> None:
+ environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.settings.user,
+ password=doris_sandbox.settings.password,
+ )
+ assert environment.get("DORIS_FE_HTTP_PORT")
+ assert environment.get("DORIS_BE_HOSTS")
+ assert environment.get("DORIS_BE_WEBSERVER_PORT")
+
+ async with _transport_client(transport, environment) as client:
+ fe_result = await client.call_tool(
+ "get_monitoring_metrics",
+ {
+ "content_type": "data",
+ "role": "fe",
+ "monitor_type": "all",
+ "priority": "all",
+ "include_raw_metrics": False,
+ },
+ )
+ assert fe_result.is_error is False
+ assert isinstance(fe_result.structured_content, dict)
+ fe_payload = fe_result.structured_content["data"]["fe"]
+ assert fe_payload["success"] is True
+ assert fe_payload["node_type"] == "fe"
+ assert fe_payload["node_info"]["host"] == doris_sandbox.settings.host
+
+ be_result = await client.call_tool(
+ "get_monitoring_metrics",
+ {
+ "content_type": "data",
+ "role": "be",
+ "monitor_type": "all",
+ "priority": "all",
+ "include_raw_metrics": False,
+ },
+ )
+ assert be_result.is_error is False
+ assert isinstance(be_result.structured_content, dict)
+ be_payload = be_result.structured_content["data"]["be"]
+ assert be_payload
+ assert all(node["success"] is True for node in be_payload)
+ assert all(node["node_type"] == "be" for node in be_payload)
+
+ recovered_result, recovered_payload = await _exec_query(
+ client,
+ "SELECT 1 AS recovered",
+ )
+ assert recovered_result.is_error is False
+ assert recovered_payload["success"] is True
+ assert recovered_payload["data"][0]["recovered"] == 1
diff --git a/test/protocol/stdio_capability_server.py
b/test/protocol/stdio_capability_server.py
index 14adf19..f510a9c 100644
--- a/test/protocol/stdio_capability_server.py
+++ b/test/protocol/stdio_capability_server.py
@@ -41,6 +41,7 @@ from doris_mcp_server.utils.analysis_tools import SQLAnalyzer
from doris_mcp_server.utils.data_governance_tools import DataGovernanceTools
from doris_mcp_server.utils.data_quality_tools import DataQualityTools
from doris_mcp_server.utils.db import QueryResult
+from doris_mcp_server.utils.monitoring_tools import DorisMonitoringTools
from doris_mcp_server.utils.security_analytics_tools import
SecurityAnalyticsTools
REQUIRED_EXTENSION = "io.apache.doris/read"
@@ -148,6 +149,27 @@ class RoleConnectionManager:
return self.connection
+class SSRFMonitoringConnectionManager:
+ def __init__(self) -> None:
+ self.config = SimpleNamespace(
+ database=SimpleNamespace(
+ host="169.254.169.254",
+ fe_http_port=80,
+ be_hosts=[],
+ be_webserver_port=8040,
+ user="root",
+ password="",
+ http_connect_timeout_seconds=0.2,
+ http_read_timeout_seconds=0.2,
+ http_total_timeout_seconds=0.2,
+ http_max_response_bytes=1024,
+ )
+ )
+
+ async def get_connection(self, session_id: str):
+ raise AssertionError("SSRF policy must reject before database access")
+
+
class RoleSecurityAnalyticsTools(SecurityAnalyticsTools):
async def _get_audit_log_data(
self,
@@ -204,6 +226,9 @@ class OneToolManager:
),
)
self.role_analyzer =
RoleSecurityAnalyticsTools(RoleConnectionManager())
+ self.monitoring_tools = DorisMonitoringTools(
+ SSRFMonitoringConnectionManager()
+ )
async def list_tools(self) -> list[Tool]:
return [
@@ -259,6 +284,17 @@ class OneToolManager:
"required": ["table_name", "columns"],
},
),
+ Tool(
+ name="get_monitoring_metrics",
+ description="Exercise the production Doris HTTP SSRF
boundary.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "content_type": {"type": "string"},
+ "role": {"type": "string"},
+ },
+ },
+ ),
]
async def call_tool(self, name: str, arguments: dict) -> str:
@@ -292,6 +328,15 @@ class OneToolManager:
return json.dumps(
await self.freshness_router._analyze_columns_tool(arguments)
)
+ if name == "get_monitoring_metrics":
+ return json.dumps(
+ await self.monitoring_tools.get_monitoring_metrics(
+ role=arguments.get("role", "fe"),
+ priority="all",
+ info_only=arguments.get("content_type") == "definitions",
+ include_raw_metrics=False,
+ )
+ )
return "{}"
diff --git a/test/protocol/test_mcp_v2_protocol.py
b/test/protocol/test_mcp_v2_protocol.py
index 1bc3002..e476f0a 100644
--- a/test/protocol/test_mcp_v2_protocol.py
+++ b/test/protocol/test_mcp_v2_protocol.py
@@ -882,6 +882,49 @@ async def
test_http_rejects_injected_sql_identifier_and_recovers():
assert recovered.json()["result"]["isError"] is False
[email protected]
+async def test_http_monitoring_rejects_metadata_endpoint_and_recovers():
+ app = create_test_server(
+ tools_manager=ProfileToolManager(),
+ ).streamable_http_app(
+ json_response=True,
+ stateless_http=True,
+ host="127.0.0.1",
+ transport_security=create_transport_security("127.0.0.1"),
+ )
+
+ async with (
+ app.router.lifespan_context(app),
+ httpx2.ASGITransport(app) as transport,
+ httpx2.AsyncClient(
+ transport=transport,
+ base_url="http://127.0.0.1:3000";,
+ ) as client,
+ ):
+ rejected = await client.post(
+ "/mcp",
+ json=modern_tool_request(
+ 1,
+ "get_monitoring_metrics",
+ {"content_type": "data", "role": "fe"},
+ ),
+ headers=modern_tool_headers("get_monitoring_metrics"),
+ )
+ assert rejected.status_code == 200
+ assert rejected.json()["result"]["isError"] is False
+ fe_result =
rejected.json()["result"]["structuredContent"]["data"]["fe"]
+ assert fe_result["success"] is False
+ assert fe_result["error_type"] == "prohibited_endpoint"
+
+ recovered = await client.post(
+ "/mcp",
+ json=modern_tool_request(2, "echo", {}),
+ headers=modern_tool_headers("echo"),
+ )
+ assert recovered.status_code == 200
+ assert recovered.json()["result"]["isError"] is False
+
+
@pytest.mark.asyncio
async def test_stdio_validates_capabilities_versions_and_process_survival():
server_script = Path(__file__).with_name("stdio_capability_server.py")
@@ -934,6 +977,7 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
"monitor_data_freshness",
"analyze_data_access_patterns",
"analyze_columns",
+ "get_monitoring_metrics",
]
secret = "stdio-secret-sec-016"
error_result = await capable.call_tool(
@@ -961,6 +1005,7 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
"monitor_data_freshness",
"analyze_data_access_patterns",
"analyze_columns",
+ "get_monitoring_metrics",
]
legacy_error = await legacy.read_resource("doris://table/missing")
assert (
@@ -1123,6 +1168,31 @@ async def
test_stdio_rejects_injected_sql_identifier_and_recovers():
assert recovered.is_error is False
[email protected]
+async def test_stdio_monitoring_rejects_metadata_endpoint_and_recovers():
+ server_script = Path(__file__).with_name("stdio_capability_server.py")
+ server_params = StdioServerParameters(
+ command=sys.executable,
+ args=[str(server_script)],
+ )
+
+ async with Client(
+ stdio_client(server_params),
+ extensions=[advertise(REQUIRED_EXTENSION)],
+ ) as modern:
+ rejected = await modern.call_tool(
+ "get_monitoring_metrics",
+ {"content_type": "data", "role": "fe"},
+ )
+ assert rejected.is_error is False
+ fe_result = rejected.structured_content["data"]["fe"]
+ assert fe_result["success"] is False
+ assert fe_result["error_type"] == "prohibited_endpoint"
+
+ recovered = await modern.call_tool("echo", {})
+ assert recovered.is_error is False
+
+
@pytest.mark.asyncio
async def test_stdio_prompt_errors_are_typed_and_process_survives():
server_script = Path(__file__).with_name("stdio_capability_server.py")
diff --git a/test/security/test_doris_http_ssrf.py
b/test/security/test_doris_http_ssrf.py
new file mode 100644
index 0000000..13978f1
--- /dev/null
+++ b/test/security/test_doris_http_ssrf.py
@@ -0,0 +1,350 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""SSRF, timeout, and response-boundary tests for Doris FE/BE HTTP."""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import asynccontextmanager, suppress
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from doris_mcp_server.utils import doris_http_client as
doris_http_client_module
+from doris_mcp_server.utils.analysis_tools import SQLAnalyzer
+from doris_mcp_server.utils.config import DorisConfig
+from doris_mcp_server.utils.doris_http_client import (
+ MAX_RESPONSE_BYTES,
+ MAX_TIMEOUT_SECONDS,
+ DorisHTTPClient,
+ DorisHTTPPolicyError,
+ DorisHTTPRequestError,
+ DorisHTTPResponseTooLarge,
+)
+from doris_mcp_server.utils.monitoring_tools import DorisMonitoringTools
+
+
+def _database_config(
+ *,
+ host: str,
+ fe_http_port: int,
+ be_hosts: list[str] | None = None,
+ be_webserver_port: int = 8040,
+ max_response_bytes: int = 4096,
+ connect_timeout: float = 1.0,
+ read_timeout: float = 1.0,
+ total_timeout: float = 1.0,
+) -> SimpleNamespace:
+ return SimpleNamespace(
+ host=host,
+ fe_http_port=fe_http_port,
+ be_hosts=be_hosts or [],
+ be_webserver_port=be_webserver_port,
+ user="root",
+ password="",
+ http_connect_timeout_seconds=connect_timeout,
+ http_read_timeout_seconds=read_timeout,
+ http_total_timeout_seconds=total_timeout,
+ http_max_response_bytes=max_response_bytes,
+ )
+
+
+def _manager(database_config: SimpleNamespace) -> SimpleNamespace:
+ return SimpleNamespace(
+ config=SimpleNamespace(database=database_config),
+ get_connection=AsyncMock(
+ side_effect=AssertionError("HTTP monitoring must not discover SQL
nodes")
+ ),
+ )
+
+
+@asynccontextmanager
+async def _http_server(
+ *,
+ status: int = 200,
+ body: bytes = b"",
+ headers: dict[str, str] | None = None,
+ delay_seconds: float = 0,
+):
+ response_headers = {
+ "Content-Length": str(len(body)),
+ "Connection": "close",
+ **(headers or {}),
+ }
+
+ async def handler(
+ reader: asyncio.StreamReader,
+ writer: asyncio.StreamWriter,
+ ) -> None:
+ try:
+ await reader.readuntil(b"\r\n\r\n")
+ if delay_seconds:
+ await asyncio.sleep(delay_seconds)
+ reason = "OK" if status == 200 else "Found"
+ head = (
+ f"HTTP/1.1 {status} {reason}\r\n"
+ + "".join(
+ f"{key}: {value}\r\n" for key, value in
response_headers.items()
+ )
+ + "\r\n"
+ ).encode()
+ writer.write(head + body)
+ await writer.drain()
+ except (BrokenPipeError, ConnectionResetError,
asyncio.IncompleteReadError):
+ pass
+ finally:
+ writer.close()
+ with suppress(Exception):
+ await writer.wait_closed()
+
+ server = await asyncio.start_server(handler, "127.0.0.1", 0)
+ try:
+ port = int(server.sockets[0].getsockname()[1])
+ yield port
+ finally:
+ server.close()
+ await server.wait_closed()
+
+
+def _http_client(
+ host: str,
+ port: int,
+ *,
+ max_response_bytes: int = 4096,
+ connect_timeout: float = 1,
+ read_timeout: float = 1,
+ total_timeout: float = 1,
+) -> DorisHTTPClient:
+ return DorisHTTPClient(
+ user="root",
+ password="",
+ allowed_endpoints={"fe": {(host, port)}, "be": set()},
+ connect_timeout_seconds=connect_timeout,
+ read_timeout_seconds=read_timeout,
+ total_timeout_seconds=total_timeout,
+ max_response_bytes=max_response_bytes,
+ )
+
+
[email protected](
+ "host",
+ [
+ "169.254.169.254",
+ "169.254.170.2",
+ "100.100.100.200",
+ "fd00:ec2::254",
+ "::ffff:100.100.100.200",
+ "metadata.google.internal",
+ ],
+)
+async def test_configured_metadata_and_link_local_endpoints_are_rejected(
+ host: str,
+) -> None:
+ client = _http_client(host, 80)
+ with pytest.raises(DorisHTTPPolicyError, match="prohibited"):
+ await client.get(role="fe", host=host, port=80, path="/metrics")
+
+
+async def test_unconfigured_endpoint_is_rejected_before_network_access() ->
None:
+ client = _http_client("127.0.0.1", 8030)
+ with pytest.raises(DorisHTTPPolicyError, match="explicitly configured"):
+ await client.get(
+ role="fe",
+ host="127.0.0.2",
+ port=8030,
+ path="/metrics",
+ )
+
+
+async def test_hostname_resolving_to_link_local_is_rejected(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ fake_loop = SimpleNamespace(
+ getaddrinfo=AsyncMock(
+ return_value=[
+ (
+ 2,
+ 1,
+ 6,
+ "",
+ ("169.254.169.254", 8030),
+ )
+ ]
+ )
+ )
+ monkeypatch.setattr(
+ doris_http_client_module.asyncio,
+ "get_running_loop",
+ lambda: fake_loop,
+ )
+ client = _http_client("doris.internal", 8030)
+ with pytest.raises(DorisHTTPPolicyError, match="prohibited"):
+ await client._resolve_addresses("doris.internal", 8030)
+ fake_loop.getaddrinfo.assert_awaited_once()
+
+
+async def test_redirects_are_not_followed() -> None:
+ async with _http_server(
+ status=302,
+ headers={"Location": "http://169.254.169.254/latest/meta-data"},
+ ) as port:
+ response = await _http_client("127.0.0.1", port).get(
+ role="fe",
+ host="127.0.0.1",
+ port=port,
+ path="/metrics",
+ )
+ assert response.status == 302
+ assert response.body == b""
+
+
+async def test_response_content_length_is_bounded_before_body_read() -> None:
+ async with _http_server(body=b"x" * 128) as port:
+ with pytest.raises(DorisHTTPResponseTooLarge):
+ await _http_client(
+ "127.0.0.1",
+ port,
+ max_response_bytes=32,
+ ).get(
+ role="fe",
+ host="127.0.0.1",
+ port=port,
+ path="/metrics",
+ )
+
+
+async def test_read_and_total_timeouts_are_enforced() -> None:
+ async with _http_server(body=b"ok", delay_seconds=0.2) as port:
+ with pytest.raises(DorisHTTPRequestError, match="timed out"):
+ await _http_client(
+ "127.0.0.1",
+ port,
+ connect_timeout=0.05,
+ read_timeout=0.05,
+ total_timeout=0.05,
+ ).get(
+ role="fe",
+ host="127.0.0.1",
+ port=port,
+ path="/metrics",
+ )
+
+
+async def test_monitoring_does_not_discover_be_http_nodes_from_sql() -> None:
+ manager = _manager(
+ _database_config(
+ host="127.0.0.1",
+ fe_http_port=8030,
+ be_hosts=[],
+ )
+ )
+ tools = DorisMonitoringTools(manager)
+ assert await tools.get_be_nodes() == []
+ manager.get_connection.assert_not_awaited()
+
+ results = await tools._get_be_metrics("all", "p0", "prometheus", False)
+ assert results == [
+ {
+ "success": False,
+ "error": ("BE HTTP metrics require explicit DORIS_BE_HOSTS
configuration"),
+ "error_type": "unconfigured_endpoint",
+ }
+ ]
+ manager.get_connection.assert_not_awaited()
+
+
+async def test_monitoring_rejects_configured_metadata_without_network() ->
None:
+ manager = _manager(
+ _database_config(
+ host="169.254.169.254",
+ fe_http_port=80,
+ )
+ )
+ result = await DorisMonitoringTools(manager)._get_fe_metrics(
+ "all",
+ "p0",
+ "prometheus",
+ False,
+ )
+ assert result["success"] is False
+ assert result["error_type"] == "prohibited_endpoint"
+
+
+async def test_monitoring_fetches_configured_loopback_metrics() -> None:
+ metrics = b"doris_fe_query_total 7\n"
+ async with _http_server(
+ body=metrics,
+ headers={"Content-Type": "text/plain"},
+ ) as port:
+ manager = _manager(
+ _database_config(
+ host="127.0.0.1",
+ fe_http_port=port,
+ )
+ )
+ result = await DorisMonitoringTools(manager).get_monitoring_metrics(
+ role="fe",
+ priority="all",
+ include_raw_metrics=True,
+ )
+ assert result["success"] is True
+ assert result["data"]["fe"]["success"] is True
+ assert result["data"]["fe"]["metrics"]["doris_fe_query_total"] == 7
+
+
+async def test_analysis_fe_http_path_uses_same_policy() -> None:
+ manager = _manager(
+ _database_config(
+ host="metadata.google.internal",
+ fe_http_port=80,
+ )
+ )
+ analyzer = SQLAnalyzer(manager)
+ assert await analyzer._get_query_id_by_trace_id("trace-id") is None
+ result = await analyzer.get_table_data_size(db_name="db",
table_name="table")
+ assert result["success"] is False
+ assert "prohibited" in result["error"].lower()
+
+
+def test_http_safety_configuration_loads_and_has_runtime_hard_caps(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ monkeypatch.setenv("DORIS_HTTP_CONNECT_TIMEOUT_SECONDS", "2")
+ monkeypatch.setenv("DORIS_HTTP_READ_TIMEOUT_SECONDS", "4")
+ monkeypatch.setenv("DORIS_HTTP_TOTAL_TIMEOUT_SECONDS", "8")
+ monkeypatch.setenv("DORIS_HTTP_MAX_RESPONSE_BYTES", "8192")
+ config = DorisConfig.from_env()
+ assert config.database.http_connect_timeout_seconds == 2
+ assert config.database.http_read_timeout_seconds == 4
+ assert config.database.http_total_timeout_seconds == 8
+ assert config.database.http_max_response_bytes == 8192
+ assert not [error for error in config.validate() if
error.startswith("Doris HTTP")]
+
+ database_config = _database_config(
+ host="127.0.0.1",
+ fe_http_port=8030,
+ connect_timeout=999,
+ read_timeout=999,
+ total_timeout=999,
+ max_response_bytes=MAX_RESPONSE_BYTES * 2,
+ )
+ client = DorisHTTPClient.from_database_config(database_config)
+ assert client.connect_timeout_seconds == MAX_TIMEOUT_SECONDS
+ assert client.read_timeout_seconds == MAX_TIMEOUT_SECONDS
+ assert client.total_timeout_seconds == MAX_TIMEOUT_SECONDS
+ assert client.max_response_bytes == MAX_RESPONSE_BYTES
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
