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 57a7291 fix: audit and harden dynamic SQL sinks (#128)
57a7291 is described below
commit 57a7291ad0c773ad5eb35d2d7d156df49302266f
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 00:49:43 2026 +0800
fix: audit and harden dynamic SQL sinks (#128)
---
doris_mcp_server/tools/resources_manager.py | 38 ++-
doris_mcp_server/utils/analysis_tools.py | 73 ++++--
doris_mcp_server/utils/data_exploration_tools.py | 152 ++++++++----
doris_mcp_server/utils/data_governance_tools.py | 73 ++++--
doris_mcp_server/utils/data_quality_tools.py | 219 +++++++++++-----
.../utils/dependency_analysis_tools.py | 6 +-
.../utils/performance_analytics_tools.py | 67 +++--
doris_mcp_server/utils/schema_extractor.py | 166 ++++++++-----
doris_mcp_server/utils/security_analytics_tools.py | 43 +++-
doris_mcp_server/utils/sql_security_utils.py | 85 ++++++-
test/integration/test_real_doris_transports.py | 60 +++++
test/protocol/stdio_capability_server.py | 35 +++
test/protocol/test_mcp_v2_protocol.py | 76 ++++++
test/security/test_sql_sink_audit.py | 276 +++++++++++++++++++++
14 files changed, 1110 insertions(+), 259 deletions(-)
diff --git a/doris_mcp_server/tools/resources_manager.py
b/doris_mcp_server/tools/resources_manager.py
index b481c36..482acb1 100644
--- a/doris_mcp_server/tools/resources_manager.py
+++ b/doris_mcp_server/tools/resources_manager.py
@@ -30,7 +30,7 @@ from mcp.types import Resource
from ..utils.db import DorisConnectionManager
from ..utils.logger import get_logger
from ..utils.redaction import redact_uri
-from ..utils.sql_security_utils import get_auth_context
+from ..utils.sql_security_utils import get_auth_context, quote_identifier
logger = get_logger(__name__)
@@ -141,9 +141,10 @@ class DorisResourcesManager:
return getattr(get_auth_context(), "auth_method", "") == "doris_oauth"
def _schema_filter(self, column_name: str, db_name: str | None) ->
tuple[str, tuple]:
+ safe_column = quote_identifier(column_name, "information schema
column")
if db_name:
- return f"{column_name} = %s", (db_name,)
- return f"{column_name} = DATABASE()", ()
+ return f"{safe_column} = %s", (db_name,)
+ return f"{safe_column} = DATABASE()", ()
def _first_row_value(self, row: Any) -> Any:
if isinstance(row, dict):
@@ -383,6 +384,8 @@ class DorisResourcesManager:
) -> list[TableMetadata]:
"""Get table metadata for a specific database or the current
database."""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
+ # SQL sink audit: schema column -> quote_identifier and database value
+ # -> bound param; fixed template -> DorisConnection.execute.
tables_query = """
SELECT
table_name,
@@ -393,7 +396,7 @@ class DorisResourcesManager:
WHERE {schema_filter}
AND table_type = 'BASE TABLE'
ORDER BY table_name
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
auth_context = get_auth_context()
result = await connection.execute(
@@ -423,6 +426,8 @@ class DorisResourcesManager:
) -> list[dict]:
"""Get column information for table"""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
+ # SQL sink audit: schema column -> quote_identifier; database/table
+ # values -> bound params; fixed template -> DorisConnection.execute.
columns_query = """
SELECT
column_name,
@@ -435,7 +440,7 @@ class DorisResourcesManager:
WHERE {schema_filter}
AND table_name = %s
ORDER BY ordinal_position
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
auth_context = get_auth_context()
result = await connection.execute(
@@ -463,6 +468,8 @@ class DorisResourcesManager:
) -> list[ViewMetadata]:
"""Get view metadata for a specific database or the current
database."""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
+ # SQL sink audit: schema column -> quote_identifier and database value
+ # -> bound param; fixed template -> DorisConnection.execute.
views_query = """
SELECT
table_name,
@@ -471,7 +478,7 @@ class DorisResourcesManager:
FROM information_schema.views
WHERE {schema_filter}
ORDER BY table_name
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
auth_context = get_auth_context()
result = await connection.execute(
@@ -496,6 +503,8 @@ class DorisResourcesManager:
"""Get detailed structure information of table"""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
# Get basic table information
+ # SQL sink audit: schema column -> quote_identifier; database/table
+ # values -> bound params; fixed template -> DorisConnection.execute.
table_info_query = """
SELECT
table_name,
@@ -506,7 +515,7 @@ class DorisResourcesManager:
FROM information_schema.tables
WHERE {schema_filter}
AND table_name = %s
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
async with self._connection_context("system") as connection:
auth_context = get_auth_context()
@@ -545,6 +554,8 @@ class DorisResourcesManager:
) -> list[dict]:
"""Get index information for table"""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
+ # SQL sink audit: schema column -> quote_identifier; database/table
+ # values -> bound params; fixed template -> DorisConnection.execute.
indexes_query = """
SELECT
index_name,
@@ -555,7 +566,7 @@ class DorisResourcesManager:
WHERE {schema_filter}
AND table_name = %s
ORDER BY index_name, seq_in_index
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
auth_context = get_auth_context()
result = await connection.execute(
@@ -568,6 +579,8 @@ class DorisResourcesManager:
async def _get_view_definition(self, view_name: str, db_name: str | None =
None) -> str:
"""Get definition information of view"""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
+ # SQL sink audit: schema column -> quote_identifier; database/view
+ # values -> bound params; fixed template -> DorisConnection.execute.
view_query = """
SELECT
table_name,
@@ -576,7 +589,7 @@ class DorisResourcesManager:
FROM information_schema.views
WHERE {schema_filter}
AND table_name = %s
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
async with self._connection_context("system") as connection:
auth_context = get_auth_context()
@@ -604,6 +617,8 @@ class DorisResourcesManager:
"""Get database statistics"""
schema_filter, schema_params = self._schema_filter("table_schema",
db_name)
# Get table statistics
+ # SQL sink audit: schema column -> quote_identifier and database value
+ # -> bound param; fixed template -> DorisConnection.execute.
table_stats_query = """
SELECT
COUNT(*) as table_count,
@@ -611,7 +626,7 @@ class DorisResourcesManager:
FROM information_schema.tables
WHERE {schema_filter}
AND table_type = 'BASE TABLE'
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
async with self._connection_context("system") as connection:
auth_context = get_auth_context()
@@ -623,11 +638,12 @@ class DorisResourcesManager:
table_stats = table_result.data[0] if table_result.data else {}
# Get view statistics
+ # SQL sink audit: same quoted schema column and bound database
value.
view_stats_query = """
SELECT COUNT(*) as view_count
FROM information_schema.views
WHERE {schema_filter}
- """.format(schema_filter=schema_filter)
+ """.format(schema_filter=schema_filter) # nosec B608
view_result = await connection.execute(
view_stats_query,
diff --git a/doris_mcp_server/utils/analysis_tools.py
b/doris_mcp_server/utils/analysis_tools.py
index ecf33a7..070e861 100644
--- a/doris_mcp_server/utils/analysis_tools.py
+++ b/doris_mcp_server/utils/analysis_tools.py
@@ -34,7 +34,8 @@ from .sql_security_utils import (
validate_identifier,
quote_identifier,
build_table_reference,
- get_auth_context
+ get_auth_context,
+ validate_integer,
)
logger = get_logger(__name__)
@@ -107,9 +108,19 @@ class TableAnalyzer:
}
# Get sample data using quoted identifier
- if include_sample and sample_size > 0:
+ safe_sample_size = validate_integer(
+ sample_size,
+ "sample size",
+ minimum=0,
+ maximum=10000,
+ )
+ if include_sample and safe_sample_size > 0:
quoted_table = quote_identifier(table_name, "table name")
- sample_sql = f"SELECT * FROM {quoted_table} LIMIT {sample_size}"
+ # SQL sink audit: caller integer -> bounded validation + quoted
+ # identifier -> DorisConnection.execute.
+ sample_sql = (
+ f"SELECT * FROM {quoted_table} LIMIT {safe_sample_size}" #
nosec B608
+ )
sample_result = await connection.execute(sample_sql,
auth_context=auth_context)
summary["sample_data"] = sample_result.data
@@ -123,20 +134,28 @@ class TableAnalyzer:
) -> Dict[str, Any]:
"""Analyze column statistics"""
try:
+ safe_table = quote_identifier(table_name, "table name")
+ safe_column = quote_identifier(column_name, "column name")
connection = await self.connection_manager.get_connection("query")
# Basic statistics
+ # SQL sink audit: MCP identifiers -> quote_identifier; label value
+ # -> bound parameter -> DorisConnection.execute.
basic_stats_sql = f"""
SELECT
- '{column_name}' as column_name,
+ %s as column_name,
COUNT(*) as total_count,
- COUNT({column_name}) as non_null_count,
- COUNT(DISTINCT {column_name}) as distinct_count
- FROM {table_name}
- """
+ COUNT({safe_column}) as non_null_count,
+ COUNT(DISTINCT {safe_column}) as distinct_count
+ FROM {safe_table}
+ """ # nosec B608
auth_context = get_auth_context()
- basic_result = await connection.execute(basic_stats_sql,
auth_context=auth_context)
+ basic_result = await connection.execute(
+ basic_stats_sql,
+ params=(column_name,),
+ auth_context=auth_context,
+ )
if not basic_result.data:
return {
"success": False,
@@ -149,16 +168,18 @@ class TableAnalyzer:
if analysis_type in ["distribution", "detailed"]:
# Data distribution analysis
+ # SQL sink audit: MCP identifiers -> quote_identifier ->
+ # DorisConnection.execute.
distribution_sql = f"""
SELECT
- {column_name} as value,
+ {safe_column} as value,
COUNT(*) as frequency
- FROM {table_name}
- WHERE {column_name} IS NOT NULL
- GROUP BY {column_name}
+ FROM {safe_table}
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY {safe_column}
ORDER BY frequency DESC
LIMIT 20
- """
+ """ # nosec B608
distribution_result = await
connection.execute(distribution_sql, auth_context=auth_context)
analysis["value_distribution"] = distribution_result.data
@@ -166,14 +187,16 @@ class TableAnalyzer:
if analysis_type == "detailed":
# Detailed statistics (for numeric types)
try:
+ # SQL sink audit: MCP identifiers -> quote_identifier ->
+ # DorisConnection.execute.
numeric_stats_sql = f"""
SELECT
- MIN({column_name}) as min_value,
- MAX({column_name}) as max_value,
- AVG({column_name}) as avg_value
- FROM {table_name}
- WHERE {column_name} IS NOT NULL
- """
+ MIN({safe_column}) as min_value,
+ MAX({safe_column}) as max_value,
+ AVG({safe_column}) as avg_value
+ FROM {safe_table}
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
numeric_result = await
connection.execute(numeric_stats_sql, auth_context=auth_context)
if numeric_result.data:
@@ -202,18 +225,22 @@ class TableAnalyzer:
connection = await self.connection_manager.get_connection("system")
# Get table basic information
- table_info_sql = f"""
+ table_info_sql = """
SELECT
table_name,
table_comment,
table_rows
FROM information_schema.tables
WHERE table_schema = DATABASE()
- AND table_name = '{table_name}'
+ AND table_name = %s
"""
auth_context = get_auth_context()
- table_result = await connection.execute(table_info_sql,
auth_context=auth_context)
+ table_result = await connection.execute(
+ table_info_sql,
+ params=(table_name,),
+ auth_context=auth_context,
+ )
if not table_result.data:
raise ValueError(f"Table {table_name} does not exist")
diff --git a/doris_mcp_server/utils/data_exploration_tools.py
b/doris_mcp_server/utils/data_exploration_tools.py
index b3e2a76..ce25aeb 100644
--- a/doris_mcp_server/utils/data_exploration_tools.py
+++ b/doris_mcp_server/utils/data_exploration_tools.py
@@ -29,6 +29,7 @@ from .logger import get_logger
from .sql_security_utils import (
SQLSecurityError,
validate_identifier,
+ validate_integer,
quote_identifier,
build_table_reference,
get_auth_context
@@ -65,7 +66,9 @@ class DataExplorationTools:
# table_name should already be validated by _build_full_table_name
auth_context = get_auth_context()
- count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}"
+ # SQL sink audit: table_name is produced by build_table_reference
+ # before this private helper reaches DorisConnection.execute.
+ count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}" #
nosec B608
result = await connection.execute(count_sql,
auth_context=auth_context)
if result.data:
@@ -102,6 +105,8 @@ class DataExplorationTools:
else:
where_conditions.append("table_schema = DATABASE()")
+ # SQL sink audit: WHERE fragments are fixed locally and all values
+ # are bound through params before DorisConnection.execute.
columns_sql = f"""
SELECT
column_name,
@@ -112,7 +117,7 @@ class DataExplorationTools:
FROM information_schema.columns
WHERE {' AND '.join(where_conditions)}
ORDER BY ordinal_position
- """
+ """ # nosec B608
result = await connection.execute(columns_sql,
params=tuple(params), auth_context=auth_context)
return result.data if result.data else []
@@ -126,7 +131,13 @@ class DataExplorationTools:
async def _determine_sampling_strategy(self, connection, table_name: str,
total_rows: int, sample_size: int) -> Dict[str, Any]:
"""Determine optimal sampling strategy based on table size"""
- if total_rows <= sample_size:
+ safe_sample_size = validate_integer(
+ sample_size,
+ "sample size",
+ minimum=0,
+ maximum=10_000_000,
+ )
+ if total_rows <= safe_sample_size:
# Use all data if table is small enough
return {
"total_rows": total_rows,
@@ -138,14 +149,19 @@ class DataExplorationTools:
}
else:
# Use random sampling for large tables
- sampling_ratio = sample_size / total_rows
+ sampling_ratio = safe_sample_size / total_rows
return {
"total_rows": total_rows,
- "sample_size": sample_size,
+ "sample_size": safe_sample_size,
"sampling_method": "random_sample",
"sampling_ratio": round(sampling_ratio, 4),
"use_sampling": True,
- "sample_table_expression": f"(SELECT * FROM {table_name} ORDER
BY RAND() LIMIT {sample_size}) as sample_table"
+ # SQL sink audit: table -> build_table_reference; sample size
->
+ # bounded integer; expression remains internal until execute.
+ "sample_table_expression": (
+ f"(SELECT * FROM {table_name} ORDER BY RAND() "
+ f"LIMIT {safe_sample_size}) as sample_table" # nosec B608
+ ),
}
def _select_analysis_columns(self, columns_info: List[Dict], include_all:
bool) -> List[Dict]:
@@ -194,18 +210,22 @@ class DataExplorationTools:
for column in numeric_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
# Basic statistics
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
stats_sql = f"""
SELECT
- COUNT({col_name}) as count,
- MIN({col_name}) as min_value,
- MAX({col_name}) as max_value,
- AVG({col_name}) as mean_value,
- STDDEV({col_name}) as std_dev
+ COUNT({safe_column}) as count,
+ MIN({safe_column}) as min_value,
+ MAX({safe_column}) as max_value,
+ AVG({safe_column}) as mean_value,
+ STDDEV({safe_column}) as std_dev
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
stats_result = await connection.execute(stats_sql,
auth_context=auth_context)
@@ -247,18 +267,22 @@ class DataExplorationTools:
async def _calculate_percentiles(self, connection, table_name: str,
col_name: str, sampling_info: Dict) -> Dict[str, float]:
"""Calculate percentiles for numeric column"""
try:
+ safe_column = quote_identifier(col_name, "column name")
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ # SQL sink audit: metadata column -> quote_identifier; table_expr
+ # originates from a safe table/sampler before
+ # DorisConnection.execute.
percentile_sql = f"""
SELECT
- PERCENTILE({col_name}, 0.25) as p25,
- PERCENTILE({col_name}, 0.50) as p50,
- PERCENTILE({col_name}, 0.75) as p75,
- PERCENTILE({col_name}, 0.90) as p90,
- PERCENTILE({col_name}, 0.95) as p95,
- PERCENTILE({col_name}, 0.99) as p99
+ PERCENTILE({safe_column}, 0.25) as p25,
+ PERCENTILE({safe_column}, 0.50) as p50,
+ PERCENTILE({safe_column}, 0.75) as p75,
+ PERCENTILE({safe_column}, 0.90) as p90,
+ PERCENTILE({safe_column}, 0.95) as p95,
+ PERCENTILE({safe_column}, 0.99) as p99
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(percentile_sql,
auth_context=auth_context)
@@ -291,17 +315,25 @@ class DataExplorationTools:
lower_bound = q1 - 1.5 * iqr
upper_bound = q3 + 1.5 * iqr
+ safe_column = quote_identifier(col_name, "column name")
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ # SQL sink audit: metadata column -> quote_identifier; numeric
+ # thresholds -> bound params; table_expr -> safe table/sampler.
outlier_sql = f"""
SELECT
COUNT(*) as total_count,
- SUM(CASE WHEN {col_name} < {lower_bound} OR {col_name} >
{upper_bound} THEN 1 ELSE 0 END) as outlier_count
+ SUM(CASE WHEN {safe_column} < %s OR {safe_column} > %s
+ THEN 1 ELSE 0 END) as outlier_count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
- result = await connection.execute(outlier_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ outlier_sql,
+ params=(lower_bound, upper_bound),
+ auth_context=auth_context,
+ )
if result.data:
data = result.data[0]
@@ -382,15 +414,17 @@ class DataExplorationTools:
for column in categorical_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
# Basic cardinality and distribution
+ # SQL sink audit: metadata column -> quote_identifier; table ->
+ # build_table_reference -> DorisConnection.execute.
cardinality_sql = f"""
SELECT
- COUNT(DISTINCT {col_name}) as cardinality,
- COUNT({col_name}) as non_null_count
+ COUNT(DISTINCT {safe_column}) as cardinality,
+ COUNT({safe_column}) as non_null_count
FROM {table_name}
- WHERE {col_name} IS NOT NULL
- {sampling_info.get('sample_query_suffix', '')}
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
cardinality_result = await connection.execute(cardinality_sql,
auth_context=auth_context)
@@ -430,17 +464,20 @@ class DataExplorationTools:
try:
# Use sample table expression if sampling is enabled
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ safe_column = quote_identifier(col_name, "column name")
+ # SQL sink audit: metadata column -> quote_identifier; table_expr
+ # originates from a safe table/sampler.
distribution_sql = f"""
SELECT
- {col_name} as value,
+ {safe_column} as value,
COUNT(*) as count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- GROUP BY {col_name}
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY {safe_column}
ORDER BY COUNT(*) DESC
LIMIT 20
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(distribution_sql,
auth_context=auth_context)
@@ -482,16 +519,20 @@ class DataExplorationTools:
for column in temporal_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
# Date range analysis
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
range_sql = f"""
SELECT
- MIN({col_name}) as earliest,
- MAX({col_name}) as latest,
- COUNT({col_name}) as non_null_count
+ MIN({safe_column}) as earliest,
+ MAX({safe_column}) as latest,
+ COUNT({safe_column}) as non_null_count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
range_result = await connection.execute(range_sql,
auth_context=auth_context)
@@ -564,16 +605,20 @@ class DataExplorationTools:
"""Analyze temporal patterns like seasonality and trends"""
try:
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ safe_column = quote_identifier(col_name, "column name")
# Weekly pattern analysis
+ # SQL sink audit: metadata column -> quote_identifier; table_expr
+ # originates from a safe table/sampler before
+ # DorisConnection.execute.
weekly_pattern_sql = f"""
SELECT
- DAYOFWEEK({col_name}) as day_of_week,
+ DAYOFWEEK({safe_column}) as day_of_week,
COUNT(*) as count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- GROUP BY DAYOFWEEK({col_name})
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY DAYOFWEEK({safe_column})
ORDER BY day_of_week
- """
+ """ # nosec B608
auth_context = get_auth_context()
weekly_result = await connection.execute(weekly_pattern_sql,
auth_context=auth_context)
@@ -586,17 +631,18 @@ class DataExplorationTools:
weekly_pattern.append(round(percentage, 3))
# Monthly trend analysis (simplified)
+ # SQL sink audit: same quoted metadata column and safe table_expr.
monthly_trend_sql = f"""
SELECT
- YEAR({col_name}) as year,
- MONTH({col_name}) as month,
+ YEAR({safe_column}) as year,
+ MONTH({safe_column}) as month,
COUNT(*) as count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- GROUP BY YEAR({col_name}), MONTH({col_name})
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY YEAR({safe_column}), MONTH({safe_column})
ORDER BY year, month
LIMIT 12
- """
+ """ # nosec B608
monthly_result = await connection.execute(monthly_trend_sql,
auth_context=auth_context)
monthly_trend = "stable" # Simplified trend analysis
@@ -675,13 +721,17 @@ class DataExplorationTools:
for column in columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
table_expr = sampling_info.get("sample_table_expression",
table_name)
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
null_sql = f"""
SELECT
COUNT(*) as total_count,
- COUNT({col_name}) as non_null_count
+ COUNT({safe_column}) as non_null_count
FROM {table_expr}
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(null_sql,
auth_context=auth_context)
@@ -768,4 +818,4 @@ class DataExplorationTools:
summary["notable_patterns"] = patterns
- return summary
\ No newline at end of file
+ return summary
diff --git a/doris_mcp_server/utils/data_governance_tools.py
b/doris_mcp_server/utils/data_governance_tools.py
index 8887769..3609a73 100644
--- a/doris_mcp_server/utils/data_governance_tools.py
+++ b/doris_mcp_server/utils/data_governance_tools.py
@@ -28,6 +28,7 @@ from .db import DorisConnectionManager
from .logger import get_logger
from .sql_security_utils import (
SQLSecurityError,
+ build_rule_predicate,
validate_identifier,
quote_identifier,
build_table_reference,
@@ -253,7 +254,9 @@ class DataGovernanceTools:
auth_context = get_auth_context()
# Try to get table row count
- count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}"
+ # SQL sink audit: table_name is produced by build_table_reference
+ # before this private helper reaches DorisConnection.execute.
+ count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}" #
nosec B608
result = await connection.execute(count_sql,
auth_context=auth_context)
if result.data:
@@ -290,6 +293,8 @@ class DataGovernanceTools:
else:
where_conditions.append("table_schema = DATABASE()")
+ # SQL sink audit: where fragments are selected locally and all
values
+ # are bound through params before DorisConnection.execute.
columns_sql = f"""
SELECT
column_name,
@@ -300,7 +305,7 @@ class DataGovernanceTools:
FROM information_schema.columns
WHERE {' AND '.join(where_conditions)}
ORDER BY ordinal_position
- """
+ """ # nosec B608
result = await connection.execute(columns_sql,
params=tuple(params), auth_context=auth_context)
return result.data if result.data else []
@@ -336,13 +341,15 @@ class DataGovernanceTools:
quoted_column = quote_identifier(column_name, "column name")
# Calculate null value statistics
+ # SQL sink audit: metadata column -> quote_identifier; table ->
+ # build_table_reference -> DorisConnection.execute.
null_sql = f"""
SELECT
COUNT(*) as total_count,
COUNT({quoted_column}) as non_null_count,
COUNT(*) - COUNT({quoted_column}) as null_count
FROM {table_name}
- """
+ """ # nosec B608
result = await connection.execute(null_sql,
auth_context=auth_context)
if result.data:
@@ -383,22 +390,24 @@ class DataGovernanceTools:
for rule in business_rules:
rule_name = rule.get("rule_name", "unknown")
- sql_condition = rule.get("sql_condition", "")
-
- if not sql_condition:
- continue
-
try:
+ predicate, predicate_params = build_rule_predicate(rule)
# Check number of records meeting conditions
+ # SQL sink audit: structured rule -> allowlisted operator,
+ # quoted identifier and bound values; table -> safe reference.
compliance_sql = f"""
SELECT
COUNT(*) as total_count,
- SUM(CASE WHEN {sql_condition} THEN 1 ELSE 0 END) as
pass_count
+ SUM(CASE WHEN {predicate} THEN 1 ELSE 0 END) as pass_count
FROM {table_name}
- """
+ """ # nosec B608
auth_context = get_auth_context()
- result = await connection.execute(compliance_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ compliance_sql,
+ params=predicate_params or None,
+ auth_context=auth_context,
+ )
if result.data:
stats = result.data[0]
pass_count = stats["pass_count"] or 0
@@ -406,7 +415,10 @@ class DataGovernanceTools:
pass_rate = pass_count / total_rows if total_rows > 0 else 0
compliance_results[rule_name] = {
- "rule_condition": sql_condition,
+ "rule_condition": {
+ "column": rule.get("column"),
+ "operator": rule.get("operator", "="),
+ },
"total_records": total_rows,
"pass_count": pass_count,
"fail_count": fail_count,
@@ -414,6 +426,12 @@ class DataGovernanceTools:
"compliance_score": round(pass_rate, 4)
}
+ except SQLSecurityError as e:
+ logger.warning(f"Rejected unsafe business rule {rule_name}:
{str(e)}")
+ compliance_results[rule_name] = {
+ "error": str(e),
+ "compliance_score": 0.0,
+ }
except Exception as e:
logger.warning(f"Failed to check business rule {rule_name}:
{str(e)}")
compliance_results[rule_name] = {
@@ -432,16 +450,19 @@ class DataGovernanceTools:
primary_key_columns = [col["column_name"] for col in columns_info
if "primary" in col.get("column_comment", "").lower()]
for pk_col in primary_key_columns:
+ safe_pk_col = quote_identifier(pk_col, "column name")
+ # SQL sink audit: metadata column -> quote_identifier; table ->
+ # build_table_reference -> DorisConnection.execute.
duplicate_sql = f"""
SELECT COUNT(*) as duplicate_count
FROM (
- SELECT {pk_col}, COUNT(*) as cnt
+ SELECT {safe_pk_col}, COUNT(*) as cnt
FROM {table_name}
- WHERE {pk_col} IS NOT NULL
- GROUP BY {pk_col}
+ WHERE {safe_pk_col} IS NOT NULL
+ GROUP BY {safe_pk_col}
HAVING COUNT(*) > 1
) t
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(duplicate_sql,
auth_context=auth_context)
@@ -531,7 +552,11 @@ class DataGovernanceTools:
safe_column = quote_identifier(column_name, "column name")
# table_name is already safe (from _build_full_table_name)
- verify_sql = f"SELECT {safe_column} FROM {table_name} LIMIT 1"
+ # SQL sink audit: column -> quote_identifier; table ->
+ # build_table_reference -> DorisConnection.execute.
+ verify_sql = (
+ f"SELECT {safe_column} FROM {table_name} LIMIT 1" # nosec B608
+ )
auth_context = get_auth_context()
await connection.execute(verify_sql, auth_context=auth_context)
return True
@@ -753,13 +778,15 @@ class DataGovernanceTools:
else:
where_clause = "table_schema = DATABASE()"
+ # SQL sink audit: where_clause has two fixed local variants and the
+ # optional database value is bound before DorisConnection.execute.
tables_sql = f"""
SELECT table_name
FROM information_schema.tables
WHERE {where_clause}
AND table_type = 'BASE TABLE'
ORDER BY table_name
- """
+ """ # nosec B608
result = await connection.execute(tables_sql, params=tuple(params)
if params else None, auth_context=auth_context)
return [row["table_name"] for row in result.data] if result.data
else []
@@ -872,10 +899,16 @@ class DataGovernanceTools:
timestamp_columns = await self._find_timestamp_columns(connection,
table_name)
if timestamp_columns:
+ safe_timestamp_column = quote_identifier(
+ timestamp_columns[0],
+ "column name",
+ )
+ # SQL sink audit: metadata column -> quote_identifier; table ->
+ # build_table_reference -> DorisConnection.execute.
max_time_sql = f"""
- SELECT MAX({timestamp_columns[0]}) as last_update
+ SELECT MAX({safe_timestamp_column}) as last_update
FROM {table_name}
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(max_time_sql,
auth_context=auth_context)
diff --git a/doris_mcp_server/utils/data_quality_tools.py
b/doris_mcp_server/utils/data_quality_tools.py
index bdcf4af..275fef0 100644
--- a/doris_mcp_server/utils/data_quality_tools.py
+++ b/doris_mcp_server/utils/data_quality_tools.py
@@ -34,6 +34,8 @@ from .config import DorisConfig
from .sql_security_utils import (
SQLSecurityError,
validate_identifier,
+ validate_integer,
+ quote_identifier,
build_table_reference,
get_auth_context
)
@@ -150,6 +152,12 @@ class DataQualityTools:
"""
try:
start_time = time.time()
+ sample_size = validate_integer(
+ sample_size,
+ "sample size",
+ minimum=0,
+ maximum=10_000_000,
+ )
logger.info(f"🔍 Analyzing columns for table: {table_name}")
logger.info(f"📊 Columns to analyze: {columns}")
logger.info(f"🎯 Analysis types: {analysis_types}")
@@ -317,7 +325,9 @@ class DataQualityTools:
auth_context = get_auth_context()
# Try to get row count
- count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}"
+ # SQL sink audit: table_name is produced by build_table_reference
+ # before this private helper reaches DorisConnection.execute.
+ count_sql = f"SELECT COUNT(*) as row_count FROM {table_name}" #
nosec B608
result = await connection.execute(count_sql,
auth_context=auth_context)
if result.data:
return {"row_count": result.data[0]["row_count"]}
@@ -388,6 +398,8 @@ class DataQualityTools:
params.append(table_name)
where_conditions.append("PARTITION_NAME IS NOT NULL")
+ # SQL sink audit: WHERE fragments are fixed locally and all values
+ # are bound through params before DorisConnection.execute.
partition_sql = f"""
SELECT
PARTITION_NAME,
@@ -397,7 +409,7 @@ class DataQualityTools:
INDEX_LENGTH
FROM information_schema.PARTITIONS
WHERE {' AND '.join(where_conditions)}
- """
+ """ # nosec B608
result = await connection.execute(partition_sql,
params=tuple(params), auth_context=auth_context)
partitions = []
@@ -454,11 +466,8 @@ class DataQualityTools:
) -> Optional[str]:
"""Get table DDL statement"""
try:
- query = (
- f"SHOW CREATE TABLE {db_name}.{table_name}"
- if db_name
- else f"SHOW CREATE TABLE {table_name}"
- )
+ safe_table_ref = build_table_reference(table_name, db_name)
+ query = f"SHOW CREATE TABLE {safe_table_ref}"
auth_context = get_auth_context()
result = await connection.execute(query, auth_context=auth_context)
if result.data:
@@ -514,11 +523,17 @@ class DataQualityTools:
async def _determine_optimized_sampling_strategy(self, connection,
table_name: str, total_rows: int, sample_size: int) -> Dict[str, Any]:
"""Determine optimized sampling strategy"""
+ safe_sample_size = validate_integer(
+ sample_size,
+ "sample size",
+ minimum=0,
+ maximum=10_000_000,
+ )
# Use thresholds from configuration
small_threshold = self.config.data_quality.small_table_threshold
medium_threshold = self.config.data_quality.medium_table_threshold
- if total_rows <= sample_size or sample_size <= 0:
+ if total_rows <= safe_sample_size or safe_sample_size <= 0:
return {
"sample_size": total_rows,
"sample_rate": 1.0,
@@ -527,7 +542,7 @@ class DataQualityTools:
"total_rows": total_rows
}
- sample_rate = sample_size / total_rows
+ sample_rate = safe_sample_size / total_rows
# Stratified sampling strategy
if total_rows <= small_threshold:
@@ -541,9 +556,13 @@ class DataQualityTools:
}
elif total_rows <= medium_threshold:
# Medium table: simple LIMIT sampling (avoid ORDER BY RAND())
- sample_table_expr = f"(SELECT * FROM {table_name} LIMIT
{sample_size}) AS sample_table"
+ # SQL sink audit: table -> build_table_reference; sample size ->
+ # bounded integer; expression remains internal until execute.
+ sample_table_expr = (
+ f"(SELECT * FROM {table_name} LIMIT {safe_sample_size}) AS
sample_table" # nosec B608
+ )
return {
- "sample_size": sample_size,
+ "sample_size": safe_sample_size,
"sample_rate": sample_rate,
"sample_table_expression": sample_table_expr,
"sampling_method": "limit_sampling",
@@ -552,15 +571,23 @@ class DataQualityTools:
else:
# Large table: use simpler sampling strategy
# For very large tables, still use LIMIT but increase sample size
to improve representativeness
- adjusted_sample_size = min(sample_size * 2, total_rows // 100) #
At most 1% sampling
- sample_table_expr = f"(SELECT * FROM {table_name} LIMIT
{adjusted_sample_size}) AS sample_table"
+ adjusted_sample_size = min(
+ safe_sample_size * 2,
+ total_rows // 100,
+ ) # At most 1% sampling
+ # SQL sink audit: table -> build_table_reference; adjusted sample
+ # size is derived from a bounded integer and DB row count before
+ # the expression reaches DorisConnection.execute.
+ sample_table_expr = (
+ f"(SELECT * FROM {table_name} LIMIT {adjusted_sample_size}) AS
sample_table" # nosec B608
+ )
return {
"sample_size": adjusted_sample_size,
"sample_rate": adjusted_sample_size / total_rows,
"sample_table_expression": sample_table_expr,
"sampling_method": "enhanced_limit_sampling",
"total_rows": total_rows,
- "original_sample_size": sample_size
+ "original_sample_size": safe_sample_size
}
async def _analyze_columns_batch(self, connection, table_name: str,
columns_info: List[Dict],
@@ -619,19 +646,29 @@ class DataQualityTools:
try:
# Build batch completeness query
select_clauses = []
+ column_aliases = {}
# First get total row count
select_clauses.append("COUNT(*) as total_rows")
# Then add statistics for each column
- for col in columns_info:
+ for index, col in enumerate(columns_info):
col_name = col["column_name"]
+ safe_column = quote_identifier(col_name, "column name")
+ non_null_alias = f"column_{index}_non_null"
+ distinct_alias = f"column_{index}_distinct"
+ column_aliases[col_name] = (non_null_alias, distinct_alias)
select_clauses.extend([
- f"COUNT({col_name}) as {col_name}_non_null",
- f"COUNT(DISTINCT {col_name}) as {col_name}_distinct"
+ f"COUNT({safe_column}) as {non_null_alias}",
+ f"COUNT(DISTINCT {safe_column}) as {distinct_alias}"
])
- batch_sql = f"SELECT {', '.join(select_clauses)} FROM {table_expr}"
+ # SQL sink audit: metadata columns -> quote_identifier; aliases are
+ # index-derived; table_expr originates from a safe table/sampler
+ # before DorisConnection.execute.
+ batch_sql = (
+ f"SELECT {', '.join(select_clauses)} FROM {table_expr}" #
nosec B608
+ )
auth_context = get_auth_context()
result = await connection.execute(batch_sql,
auth_context=auth_context)
@@ -644,8 +681,9 @@ class DataQualityTools:
for col in columns_info:
col_name = col["column_name"]
- non_null = row[f"{col_name}_non_null"]
- distinct = row[f"{col_name}_distinct"]
+ non_null_alias, distinct_alias = column_aliases[col_name]
+ non_null = row[non_null_alias]
+ distinct = row[distinct_alias]
null_count = total_rows - non_null
null_rate = null_count / total_rows if total_rows > 0 else 0
@@ -705,16 +743,30 @@ class DataQualityTools:
"""Batch numeric distribution analysis"""
try:
select_clauses = []
- for col in numeric_columns:
+ column_aliases = {}
+ for index, col in enumerate(numeric_columns):
col_name = col["column_name"]
+ safe_column = quote_identifier(col_name, "column name")
+ aliases = {
+ "min": f"column_{index}_min",
+ "max": f"column_{index}_max",
+ "avg": f"column_{index}_avg",
+ "stddev": f"column_{index}_stddev",
+ }
+ column_aliases[col_name] = aliases
select_clauses.extend([
- f"MIN({col_name}) as {col_name}_min",
- f"MAX({col_name}) as {col_name}_max",
- f"AVG({col_name}) as {col_name}_avg",
- f"STDDEV({col_name}) as {col_name}_stddev"
+ f"MIN({safe_column}) as {aliases['min']}",
+ f"MAX({safe_column}) as {aliases['max']}",
+ f"AVG({safe_column}) as {aliases['avg']}",
+ f"STDDEV({safe_column}) as {aliases['stddev']}"
])
- batch_sql = f"SELECT {', '.join(select_clauses)} FROM {table_expr}"
+ # SQL sink audit: metadata columns -> quote_identifier; aliases are
+ # index-derived; table_expr originates from a safe table/sampler
+ # before DorisConnection.execute.
+ batch_sql = (
+ f"SELECT {', '.join(select_clauses)} FROM {table_expr}" #
nosec B608
+ )
auth_context = get_auth_context()
result = await connection.execute(batch_sql,
auth_context=auth_context)
@@ -726,12 +778,17 @@ class DataQualityTools:
for col in numeric_columns:
col_name = col["column_name"]
+ aliases = column_aliases[col_name]
numeric_results[col_name] = {
"data_type": "numeric",
- "min_value": row[f"{col_name}_min"],
- "max_value": row[f"{col_name}_max"],
- "mean": round(float(row[f"{col_name}_avg"]), 4) if
row[f"{col_name}_avg"] is not None else None,
- "std_dev": round(float(row[f"{col_name}_stddev"]), 4) if
row[f"{col_name}_stddev"] is not None else None
+ "min_value": row[aliases["min"]],
+ "max_value": row[aliases["max"]],
+ "mean": round(float(row[aliases["avg"]]), 4)
+ if row[aliases["avg"]] is not None
+ else None,
+ "std_dev": round(float(row[aliases["stddev"]]), 4)
+ if row[aliases["stddev"]] is not None
+ else None,
}
return numeric_results
@@ -748,15 +805,19 @@ class DataQualityTools:
for col in categorical_columns:
col_name = col["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
# Get top 10 most frequent values
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
freq_sql = f"""
- SELECT {col_name}, COUNT(*) as frequency
+ SELECT {safe_column}, COUNT(*) as frequency
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- GROUP BY {col_name}
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY {safe_column}
ORDER BY frequency DESC
LIMIT 10
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(freq_sql,
auth_context=auth_context)
@@ -780,17 +841,29 @@ class DataQualityTools:
"""Batch temporal distribution analysis"""
try:
select_clauses = []
- for col in temporal_columns:
+ column_aliases = {}
+ for index, col in enumerate(temporal_columns):
col_name = col["column_name"]
+ safe_column = quote_identifier(col_name, "column name")
+ aliases = {
+ "min": f"column_{index}_min",
+ "max": f"column_{index}_max",
+ }
+ column_aliases[col_name] = aliases
select_clauses.extend([
- f"MIN({col_name}) as {col_name}_min",
- f"MAX({col_name}) as {col_name}_max"
+ f"MIN({safe_column}) as {aliases['min']}",
+ f"MAX({safe_column}) as {aliases['max']}",
])
if not select_clauses:
return {}
- batch_sql = f"SELECT {', '.join(select_clauses)} FROM {table_expr}"
+ # SQL sink audit: metadata columns -> quote_identifier; aliases are
+ # index-derived; table_expr originates from a safe table/sampler
+ # before DorisConnection.execute.
+ batch_sql = (
+ f"SELECT {', '.join(select_clauses)} FROM {table_expr}" #
nosec B608
+ )
auth_context = get_auth_context()
result = await connection.execute(batch_sql,
auth_context=auth_context)
@@ -802,10 +875,11 @@ class DataQualityTools:
for col in temporal_columns:
col_name = col["column_name"]
+ aliases = column_aliases[col_name]
temporal_results[col_name] = {
"data_type": "temporal",
- "min_value": row[f"{col_name}_min"],
- "max_value": row[f"{col_name}_max"]
+ "min_value": row[aliases["min"]],
+ "max_value": row[aliases["max"]],
}
return temporal_results
@@ -826,14 +900,18 @@ class DataQualityTools:
logger.info(f" 📊 [{i}/{len(columns_info)}] Analyzing column:
{col_name}")
try:
+ safe_column = quote_identifier(col_name, "column name")
# Completeness analysis SQL
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
completeness_sql = f"""
SELECT
COUNT(*) as total_count,
- COUNT({col_name}) as non_null_count,
- COUNT(*) - COUNT({col_name}) as null_count
+ COUNT({safe_column}) as non_null_count,
+ COUNT(*) - COUNT({safe_column}) as null_count
FROM {table_expr}
- """
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(completeness_sql,
auth_context=auth_context)
@@ -951,16 +1029,20 @@ class DataQualityTools:
for column in numeric_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
stats_sql = f"""
SELECT
- COUNT({col_name}) as non_null_count,
- MIN({col_name}) as min_value,
- MAX({col_name}) as max_value,
- AVG({col_name}) as mean_value,
- STDDEV({col_name}) as std_dev
+ COUNT({safe_column}) as non_null_count,
+ MIN({safe_column}) as min_value,
+ MAX({safe_column}) as max_value,
+ AVG({safe_column}) as mean_value,
+ STDDEV({safe_column}) as std_dev
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(stats_sql,
auth_context=auth_context)
@@ -993,14 +1075,18 @@ class DataQualityTools:
for column in categorical_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
# Basic statistics
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
cardinality_sql = f"""
SELECT
- COUNT(DISTINCT {col_name}) as cardinality,
- COUNT({col_name}) as non_null_count
+ COUNT(DISTINCT {safe_column}) as cardinality,
+ COUNT({safe_column}) as non_null_count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
cardinality_result = await connection.execute(cardinality_sql,
auth_context=auth_context)
@@ -1018,14 +1104,17 @@ class DataQualityTools:
# If cardinality is not too large, get distribution of top
values
if cardinality <= 50 and detailed_response:
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler
before
+ # DorisConnection.execute.
top_values_sql = f"""
- SELECT {col_name}, COUNT(*) as count
+ SELECT {safe_column}, COUNT(*) as count
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- GROUP BY {col_name}
+ WHERE {safe_column} IS NOT NULL
+ GROUP BY {safe_column}
ORDER BY COUNT(*) DESC
LIMIT 10
- """
+ """ # nosec B608
top_values_result = await
connection.execute(top_values_sql, auth_context=auth_context)
if top_values_result.data:
@@ -1047,14 +1136,18 @@ class DataQualityTools:
for column in temporal_columns:
col_name = column["column_name"]
try:
+ safe_column = quote_identifier(col_name, "column name")
+ # SQL sink audit: metadata column -> quote_identifier;
+ # table_expr originates from a safe table/sampler before
+ # DorisConnection.execute.
stats_sql = f"""
SELECT
- COUNT({col_name}) as non_null_count,
- MIN({col_name}) as min_date,
- MAX({col_name}) as max_date
+ COUNT({safe_column}) as non_null_count,
+ MIN({safe_column}) as min_date,
+ MAX({safe_column}) as max_date
FROM {table_expr}
- WHERE {col_name} IS NOT NULL
- """
+ WHERE {safe_column} IS NOT NULL
+ """ # nosec B608
auth_context = get_auth_context()
result = await connection.execute(stats_sql,
auth_context=auth_context)
diff --git a/doris_mcp_server/utils/dependency_analysis_tools.py
b/doris_mcp_server/utils/dependency_analysis_tools.py
index feb55a3..34d6a95 100644
--- a/doris_mcp_server/utils/dependency_analysis_tools.py
+++ b/doris_mcp_server/utils/dependency_analysis_tools.py
@@ -151,6 +151,8 @@ class DependencyAnalysisTools:
where_conditions.append(f"table_type IN ({','.join(table_types)})")
+ # SQL sink audit: database value -> bound param; table-type and
+ # WHERE fragments are fixed local variants ->
DorisConnection.execute.
metadata_sql = f"""
SELECT
table_schema as schema_name,
@@ -162,7 +164,7 @@ class DependencyAnalysisTools:
FROM information_schema.tables
WHERE {' AND '.join(where_conditions)}
ORDER BY table_schema, table_name
- """
+ """ # nosec B608
# SECURITY FIX: Get auth_context and pass to execute for security
validation
auth_context = get_auth_context()
@@ -1022,4 +1024,4 @@ class DependencyAnalysisTools:
"action": "Review and implement suggested optimizations
for better maintainability"
})
- return recommendations
\ No newline at end of file
+ return recommendations
diff --git a/doris_mcp_server/utils/performance_analytics_tools.py
b/doris_mcp_server/utils/performance_analytics_tools.py
index a9e4a2a..1d15dec 100644
--- a/doris_mcp_server/utils/performance_analytics_tools.py
+++ b/doris_mcp_server/utils/performance_analytics_tools.py
@@ -30,6 +30,7 @@ from .logger import get_logger
from .sql_security_utils import (
SQLSecurityError,
validate_identifier,
+ validate_integer,
quote_identifier,
build_table_reference,
get_auth_context
@@ -65,6 +66,14 @@ class PerformanceAnalyticsTools:
Slow query analysis results
"""
try:
+ days = validate_integer(days, "days", minimum=1, maximum=3650)
+ top_n = validate_integer(top_n, "top N", minimum=1, maximum=1000)
+ min_execution_time_ms = validate_integer(
+ min_execution_time_ms,
+ "minimum execution time",
+ minimum=0,
+ maximum=86_400_000,
+ )
start_time = time.time()
connection = await self.connection_manager.get_connection("query")
@@ -140,6 +149,7 @@ class PerformanceAnalyticsTools:
Resource growth analysis results
"""
try:
+ days = validate_integer(days, "days", minimum=1, maximum=3650)
start_time = time.time()
connection = await self.connection_manager.get_connection("query")
@@ -220,7 +230,7 @@ class PerformanceAnalyticsTools:
start_date = datetime.now() - timedelta(days=days)
# Get daily query counts from audit logs
- query_volume_sql = f"""
+ query_volume_sql = """
SELECT
DATE(`time`) as query_date,
COUNT(*) as total_queries,
@@ -229,7 +239,7 @@ class PerformanceAnalyticsTools:
SUM(`scan_bytes`) as total_scan_bytes,
SUM(`scan_rows`) as total_scan_rows
FROM internal.__internal_schema.audit_log
- WHERE `time` >= '{start_date.strftime('%Y-%m-%d %H:%M:%S')}'
+ WHERE `time` >= %s
AND `stmt` IS NOT NULL
AND `stmt` != ''
GROUP BY DATE(`time`)
@@ -237,7 +247,11 @@ class PerformanceAnalyticsTools:
"""
auth_context = get_auth_context()
- result = await connection.execute(query_volume_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ query_volume_sql,
+ params=(start_date,),
+ auth_context=auth_context,
+ )
daily_data = result.data if result.data else []
if not daily_data:
@@ -298,14 +312,14 @@ class PerformanceAnalyticsTools:
start_date = datetime.now() - timedelta(days=days)
# Get daily user activity from audit logs
- user_activity_sql = f"""
+ user_activity_sql = """
SELECT
DATE(`time`) as activity_date,
COUNT(DISTINCT `user`) as daily_active_users,
COUNT(*) as total_queries,
COUNT(DISTINCT `client_ip`) as unique_ips
FROM internal.__internal_schema.audit_log
- WHERE `time` >= '{start_date.strftime('%Y-%m-%d %H:%M:%S')}'
+ WHERE `time` >= %s
AND `stmt` IS NOT NULL
AND `stmt` != ''
GROUP BY DATE(`time`)
@@ -313,7 +327,11 @@ class PerformanceAnalyticsTools:
"""
auth_context = get_auth_context()
- result = await connection.execute(user_activity_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ user_activity_sql,
+ params=(start_date,),
+ auth_context=auth_context,
+ )
daily_data = result.data if result.data else []
if not daily_data:
@@ -369,7 +387,7 @@ class PerformanceAnalyticsTools:
try:
start_date = datetime.now() - timedelta(days=days)
- slow_query_sql = f"""
+ slow_query_sql = """
SELECT
`user` as user_name,
`client_ip` as host,
@@ -380,8 +398,8 @@ class PerformanceAnalyticsTools:
`scan_rows` as scan_rows,
`return_rows` as return_rows
FROM internal.__internal_schema.audit_log
- WHERE `time` >= '{start_date.strftime('%Y-%m-%d %H:%M:%S')}'
- AND `query_time` >= {min_execution_time_ms}
+ WHERE `time` >= %s
+ AND `query_time` >= %s
AND `stmt` IS NOT NULL
AND `stmt` != ''
AND `stmt` NOT LIKE '%__internal_schema%'
@@ -393,7 +411,11 @@ class PerformanceAnalyticsTools:
"""
auth_context = get_auth_context()
- result = await connection.execute(slow_query_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ slow_query_sql,
+ params=(start_date, min_execution_time_ms),
+ auth_context=auth_context,
+ )
return result.data if result.data else []
except Exception as e:
@@ -645,7 +667,17 @@ class PerformanceAnalyticsTools:
for table_info in selected_tables["tables"]:
table_name = table_info["table_name"]
schema_name = table_info["schema_name"]
- full_table_name = f"{schema_name}.{table_name}" if schema_name
else table_name
+ try:
+ full_table_name = build_table_reference(
+ table_name,
+ schema_name or None,
+ )
+ except SQLSecurityError as exc:
+ logger.warning(
+ "Skipping table with unsafe metadata identifier: %s",
+ exc,
+ )
+ continue
logger.info(f"🔍 Analyzing table: {full_table_name}
({table_info['size_mb']:.1f}MB)")
@@ -1190,7 +1222,11 @@ class PerformanceAnalyticsTools:
# If information_schema has no data, try COUNT query
# full_table_name should already be validated by caller using
build_table_reference
- count_sql = f"SELECT COUNT(*) as rows FROM {full_table_name}"
+ # SQL sink audit: full_table_name is produced by
+ # build_table_reference before DorisConnection.execute.
+ count_sql = (
+ f"SELECT COUNT(*) as rows FROM {full_table_name}" # nosec B608
+ )
count_result = await connection.execute(count_sql,
auth_context=auth_context)
if count_result.data:
return {
@@ -1304,6 +1340,9 @@ class PerformanceAnalyticsTools:
quoted_time_column = quote_identifier(time_column, "column name")
# Aggregate data by date (full_table_name should be validated by
caller)
+ # SQL sink audit: metadata column -> quote_identifier; table ->
+ # build_table_reference; days -> bound param before
+ # DorisConnection.execute.
growth_sql = f"""
SELECT
DATE({quoted_time_column}) as date,
@@ -1314,7 +1353,7 @@ class PerformanceAnalyticsTools:
AND {quoted_time_column} IS NOT NULL
GROUP BY DATE({quoted_time_column})
ORDER BY date DESC
- """
+ """ # nosec B608
result = await connection.execute(growth_sql, params=(days,),
auth_context=auth_context)
if not result.data:
@@ -1807,4 +1846,4 @@ class PerformanceAnalyticsTools:
"action": "Review partitioning strategies and add
appropriate indexes"
})
- return recommendations
\ No newline at end of file
+ return recommendations
diff --git a/doris_mcp_server/utils/schema_extractor.py
b/doris_mcp_server/utils/schema_extractor.py
index e6da0bb..5454658 100644
--- a/doris_mcp_server/utils/schema_extractor.py
+++ b/doris_mcp_server/utils/schema_extractor.py
@@ -35,7 +35,8 @@ from .logger import get_logger
from .sql_security_utils import (
SQLSecurityError,
validate_identifier,
- quote_identifier
+ validate_integer,
+ quote_identifier,
)
# Configure logging
@@ -195,14 +196,19 @@ class MetadataExtractor:
effective_db = db_name or self.db_name
effective_catalog = catalog_name or self.catalog_name
- query = f"""
+ query = """
SELECT 1 AS TABLE_VISIBLE
FROM information_schema.tables
- WHERE TABLE_SCHEMA = '{effective_db}'
- AND TABLE_NAME = '{table_name}'
+ WHERE TABLE_SCHEMA = %s
+ AND TABLE_NAME = %s
LIMIT 1
"""
- result = await self._execute_query_with_catalog_async(query,
effective_db, effective_catalog)
+ result = await self._execute_query_with_catalog_async(
+ query,
+ effective_db,
+ effective_catalog,
+ params=(effective_db, table_name),
+ )
if not result or not result[0]:
raise self._doris_oauth_table_not_visible_error(table_name,
effective_db, effective_catalog)
@@ -368,6 +374,14 @@ class MetadataExtractor:
if not db_name:
logger.warning("Database name not specified")
return []
+
+ try:
+ validate_identifier(db_name, "database name")
+ if effective_catalog:
+ validate_identifier(effective_catalog, "catalog name")
+ except SQLSecurityError as exc:
+ logger.warning(f"Invalid identifier rejected: {exc}")
+ return []
cache_key = f"tables_{effective_catalog or 'default'}_{db_name}"
if cache_key in self.metadata_cache and (datetime.now() -
self.metadata_cache_time.get(cache_key, datetime.min)).total_seconds() <
self.cache_ttl:
@@ -375,6 +389,8 @@ class MetadataExtractor:
try:
# Use information_schema.tables table to get table list
+ # SQL sink audit: deprecated sync path validates both identifiers
+ # before the fixed metadata query reaches its legacy executor.
query = f"""
SELECT
TABLE_NAME
@@ -383,7 +399,7 @@ class MetadataExtractor:
WHERE
TABLE_SCHEMA = '{db_name}'
AND TABLE_TYPE = 'BASE TABLE'
- """
+ """ # nosec B608
result = self._execute_query_with_catalog(query, db_name,
effective_catalog)
logger.info(
@@ -572,7 +588,7 @@ class MetadataExtractor:
try:
# Use information_schema.columns table to get table schema (async)
- query = f"""
+ query = """
SELECT
COLUMN_NAME,
DATA_TYPE,
@@ -585,13 +601,18 @@ class MetadataExtractor:
FROM
information_schema.columns
WHERE
- TABLE_SCHEMA = '{db_name}'
- AND TABLE_NAME = '{table_name}'
+ TABLE_SCHEMA = %s
+ AND TABLE_NAME = %s
ORDER BY
ORDINAL_POSITION
"""
- result = await self._execute_query_with_catalog_async(query,
db_name, effective_catalog)
+ result = await self._execute_query_with_catalog_async(
+ query,
+ db_name,
+ effective_catalog,
+ params=(db_name, table_name),
+ )
if not result:
logger.warning(f"Table {effective_catalog or
'default'}.{db_name}.{table_name} does not exist or has no columns")
@@ -626,17 +647,20 @@ class MetadataExtractor:
# Get table type information (async)
try:
- table_type_query = f"""
+ table_type_query = """
SELECT
TABLE_TYPE,
ENGINE
FROM
information_schema.tables
WHERE
- TABLE_SCHEMA = '{db_name}'
- AND TABLE_NAME = '{table_name}'
+ TABLE_SCHEMA = %s
+ AND TABLE_NAME = %s
"""
- table_type_result = await
self._execute_query_async(table_type_query)
+ table_type_result = await self._execute_query_async(
+ table_type_query,
+ params=(db_name, table_name),
+ )
if table_type_result:
schema["table_type"] =
table_type_result[0].get("TABLE_TYPE", "")
schema["engine"] = table_type_result[0].get("ENGINE", "")
@@ -687,6 +711,8 @@ class MetadataExtractor:
try:
# Use information_schema.tables table to get table comment
+ # SQL sink audit: deprecated sync path validates table, database
+ # and catalog identifiers before its legacy executor.
query = f"""
SELECT
TABLE_COMMENT
@@ -695,7 +721,7 @@ class MetadataExtractor:
WHERE
TABLE_SCHEMA = '{db_name}'
AND TABLE_NAME = '{table_name}'
- """
+ """ # nosec B608
result = self._execute_query_with_catalog(query, db_name,
effective_catalog)
@@ -748,6 +774,8 @@ class MetadataExtractor:
try:
# Use information_schema.columns table to get column comments
+ # SQL sink audit: deprecated sync path validates table, database
+ # and catalog identifiers before its legacy executor.
query = f"""
SELECT
COLUMN_NAME,
@@ -759,7 +787,7 @@ class MetadataExtractor:
AND TABLE_NAME = '{table_name}'
ORDER BY
ORDINAL_POSITION
- """
+ """ # nosec B608
result = self._execute_query_with_catalog(query, db_name,
effective_catalog)
@@ -931,20 +959,6 @@ class MetadataExtractor:
pd.DataFrame: Audit log DataFrame
"""
try:
- start_date = (datetime.now() -
timedelta(days=days)).strftime('%Y-%m-%d')
-
- query = f"""
- SELECT client_ip, user, db, time, stmt_id, stmt, state, error_code
- FROM `__internal_schema`.`audit_log`
- WHERE `time` >= '{start_date}'
- AND state = 'EOF' AND error_code = 0
- AND `stmt` NOT LIKE 'SHOW%'
- AND `stmt` NOT LIKE 'DESC%'
- AND `stmt` NOT LIKE 'EXPLAIN%'
- AND `stmt` NOT LIKE 'SELECT 1%'
- ORDER BY time DESC
- LIMIT {limit}
- """
# Deprecated sync path removed; this method is deprecated overall
df = pd.DataFrame()
return df
@@ -1241,20 +1255,6 @@ class MetadataExtractor:
Dict: Partition information
"""
try:
- # Get partition information
- query = f"""
- SELECT
- PARTITION_NAME,
- PARTITION_EXPRESSION,
- PARTITION_DESCRIPTION,
- TABLE_ROWS
- FROM
- information_schema.partitions
- WHERE
- TABLE_SCHEMA = '{db_name}'
- AND TABLE_NAME = '{table_name}'
- """
-
# Deprecated sync path removed
partitions = []
@@ -1281,7 +1281,13 @@ class MetadataExtractor:
# Removed sync _execute_query_with_catalog; use async variant instead
- async def _execute_query_with_catalog_async(self, query: str, db_name: str
= None, catalog_name: str = None):
+ async def _execute_query_with_catalog_async(
+ self,
+ query: str,
+ db_name: str = None,
+ catalog_name: str = None,
+ params: tuple | None = None,
+ ):
"""
Async version of _execute_query_with_catalog to avoid cross-event-loop
issues.
@@ -1291,19 +1297,33 @@ class MetadataExtractor:
"""
try:
if catalog_name and 'information_schema' in query.lower():
- modified_query = query.replace('information_schema',
f'{catalog_name}.information_schema')
+ safe_catalog = quote_identifier(catalog_name, "catalog name")
+ modified_query = query.replace(
+ 'information_schema',
+ f'{safe_catalog}.information_schema',
+ )
logger.info(
"Prepared catalog-qualified query for %s",
catalog_name,
)
- return await self._execute_query_async(modified_query, db_name)
+ return await self._execute_query_async(
+ modified_query,
+ db_name,
+ params=params,
+ )
else:
- return await self._execute_query_async(query, db_name)
+ return await self._execute_query_async(query, db_name,
params=params)
except Exception as e:
logger.error(f"Error executing async query with catalog: {str(e)}")
raise
- async def _execute_query_async(self, query: str, db_name: str = None,
return_dataframe: bool = False):
+ async def _execute_query_async(
+ self,
+ query: str,
+ db_name: str = None,
+ return_dataframe: bool = False,
+ params: tuple | None = None,
+ ):
"""
Execute database query asynchronously
@@ -1319,7 +1339,12 @@ class MetadataExtractor:
try:
if self.connection_manager:
# Use the injected connection manager directly (async)
- result = await
self.connection_manager.execute_query(self._session_id, query, None,
auth_context)
+ result = await self.connection_manager.execute_query(
+ self._session_id,
+ query,
+ params,
+ auth_context,
+ )
# Extract data from QueryResult
if hasattr(result, 'data'):
@@ -1549,17 +1574,22 @@ class MetadataExtractor:
logger.warning(f"Invalid identifier rejected: {e}")
return ""
- query = f"""
+ query = """
SELECT
TABLE_COMMENT
FROM
information_schema.tables
WHERE
- TABLE_SCHEMA = '{effective_db}'
- AND TABLE_NAME = '{table_name}'
+ TABLE_SCHEMA = %s
+ AND TABLE_NAME = %s
"""
- result = await self._execute_query_with_catalog_async(query,
effective_db, effective_catalog)
+ result = await self._execute_query_with_catalog_async(
+ query,
+ effective_db,
+ effective_catalog,
+ params=(effective_db, table_name),
+ )
if not result or not result[0]:
self._raise_if_doris_oauth_table_not_visible(table_name,
effective_db, effective_catalog)
return ""
@@ -1584,20 +1614,25 @@ class MetadataExtractor:
logger.warning(f"Invalid identifier rejected: {e}")
return {}
- query = f"""
+ query = """
SELECT
COLUMN_NAME,
COLUMN_COMMENT
FROM
information_schema.columns
WHERE
- TABLE_SCHEMA = '{effective_db}'
- AND TABLE_NAME = '{table_name}'
+ TABLE_SCHEMA = %s
+ AND TABLE_NAME = %s
ORDER BY
ORDINAL_POSITION
"""
- rows = await self._execute_query_with_catalog_async(query,
effective_db, effective_catalog)
+ rows = await self._execute_query_with_catalog_async(
+ query,
+ effective_db,
+ effective_catalog,
+ params=(effective_db, table_name),
+ )
if not rows:
await self._ensure_table_visible_for_doris_oauth_metadata(
table_name,
@@ -1686,20 +1721,25 @@ class MetadataExtractor:
async def get_recent_audit_logs_async(self, days: int = 7, limit: int =
100):
"""Async version: get recent audit logs and return a pandas
DataFrame."""
try:
- start_date = (datetime.now() -
timedelta(days=days)).strftime('%Y-%m-%d')
- query = f"""
+ safe_days = validate_integer(days, "days", minimum=1, maximum=3650)
+ safe_limit = validate_integer(limit, "limit", minimum=1,
maximum=10000)
+ start_date = datetime.now() - timedelta(days=safe_days)
+ query = """
SELECT client_ip, user, db, time, stmt_id, stmt, state, error_code
FROM `__internal_schema`.`audit_log`
- WHERE `time` >= '{start_date}'
+ WHERE `time` >= %s
AND state = 'EOF' AND error_code = 0
AND `stmt` NOT LIKE 'SHOW%'
AND `stmt` NOT LIKE 'DESC%'
AND `stmt` NOT LIKE 'EXPLAIN%'
AND `stmt` NOT LIKE 'SELECT 1%'
ORDER BY time DESC
- LIMIT {limit}
+ LIMIT %s
"""
- rows = await self._execute_query_async(query)
+ rows = await self._execute_query_async(
+ query,
+ params=(start_date, safe_limit),
+ )
import pandas as pd
return pd.DataFrame(rows or [])
except Exception as e:
diff --git a/doris_mcp_server/utils/security_analytics_tools.py
b/doris_mcp_server/utils/security_analytics_tools.py
index abbf78d..619fcdc 100644
--- a/doris_mcp_server/utils/security_analytics_tools.py
+++ b/doris_mcp_server/utils/security_analytics_tools.py
@@ -26,7 +26,7 @@ from collections import Counter, defaultdict
from .db import DorisConnectionManager
from .logger import get_logger
-from .sql_security_utils import get_auth_context
+from .sql_security_utils import get_auth_context, validate_integer
logger = get_logger(__name__)
@@ -57,6 +57,13 @@ class SecurityAnalyticsTools:
"""
connection = None
try:
+ days = validate_integer(days, "days", minimum=1, maximum=3650)
+ min_query_threshold = validate_integer(
+ min_query_threshold,
+ "minimum query threshold",
+ minimum=0,
+ maximum=1_000_000,
+ )
start_time = time.time()
# 🚀 PROGRESS: Initialize security analysis
@@ -176,11 +183,15 @@ class SecurityAnalyticsTools:
try:
# System users filter
system_user_filter = ""
+ params = [start_date, end_date]
if not include_system_users:
system_users = ['root', 'admin', 'system', 'doris',
'information_schema']
- user_list = ','.join([f'"{user}"' for user in system_users])
- system_user_filter = f"AND `user` NOT IN ({user_list})"
+ placeholders = ", ".join("%s" for _ in system_users)
+ system_user_filter = f"AND `user` NOT IN ({placeholders})"
+ params.extend(system_users)
+ # SQL sink audit: filter has only fixed local variants; date/user
+ # values are bound through params before DorisConnection.execute.
audit_sql = f"""
SELECT
`user` as user_name,
@@ -193,24 +204,30 @@ class SecurityAnalyticsTools:
`return_rows` as return_rows,
`query_time` as execution_time_ms
FROM internal.__internal_schema.audit_log
- WHERE `time` >= '{start_date.strftime('%Y-%m-%d %H:%M:%S')}'
- AND `time` <= '{end_date.strftime('%Y-%m-%d %H:%M:%S')}'
+ WHERE `time` >= %s
+ AND `time` <= %s
AND `stmt` IS NOT NULL
AND `stmt` != ''
{system_user_filter}
ORDER BY `time` DESC
LIMIT 10000
- """
+ """ # nosec B608
# SECURITY FIX: Pass auth_context to execute
auth_context = get_auth_context()
- result = await connection.execute(audit_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ audit_sql,
+ params=tuple(params),
+ auth_context=auth_context,
+ )
return result.data if result.data else []
except Exception as e:
logger.warning(f"Failed to get audit log data: {str(e)}")
# Try alternative method without detailed metrics
try:
+ # SQL sink audit: same fixed filter and bound values as the
+ # primary audit query.
simple_audit_sql = f"""
SELECT
`user` as user_name,
@@ -219,16 +236,20 @@ class SecurityAnalyticsTools:
`stmt` as sql_statement,
`state` as query_status
FROM internal.__internal_schema.audit_log
- WHERE `time` >= '{start_date.strftime('%Y-%m-%d %H:%M:%S')}'
- AND `time` <= '{end_date.strftime('%Y-%m-%d %H:%M:%S')}'
+ WHERE `time` >= %s
+ AND `time` <= %s
AND `stmt` IS NOT NULL
{system_user_filter}
ORDER BY `time` DESC
LIMIT 10000
- """
+ """ # nosec B608
auth_context = get_auth_context()
- result = await connection.execute(simple_audit_sql,
auth_context=auth_context)
+ result = await connection.execute(
+ simple_audit_sql,
+ params=tuple(params),
+ auth_context=auth_context,
+ )
return result.data if result.data else []
except Exception as e2:
diff --git a/doris_mcp_server/utils/sql_security_utils.py
b/doris_mcp_server/utils/sql_security_utils.py
index 5813daf..bddc27c 100644
--- a/doris_mcp_server/utils/sql_security_utils.py
+++ b/doris_mcp_server/utils/sql_security_utils.py
@@ -22,7 +22,8 @@ to prevent SQL injection attacks.
"""
import re
-from typing import Optional, Tuple, List, Any
+from collections.abc import Mapping, Sequence
+from typing import Optional, Tuple
from .logger import get_logger
from .security import (
@@ -67,6 +68,19 @@ class SQLSecurityUtils:
'DISTINCT', 'INTO', 'VALUES', 'SET', 'DEFAULT', 'PRIMARY', 'KEY',
'FOREIGN', 'REFERENCES', 'CHECK', 'UNIQUE', 'CONSTRAINT'
}
+
+ RULE_OPERATORS = {
+ "=",
+ "!=",
+ "<>",
+ "<",
+ ">",
+ "<=",
+ ">=",
+ "LIKE",
+ "NOT LIKE",
+ }
+ NULL_RULE_OPERATORS = {"IS NULL", "IS NOT NULL"}
@classmethod
def validate_identifier(cls, name: str, identifier_type: str =
"identifier") -> str:
@@ -256,6 +270,73 @@ class SQLSecurityUtils:
return f"{quoted_column} {operator} %s", True
else:
return f"{quoted_column} {operator}", False
+
+ @staticmethod
+ def validate_integer(
+ value: object,
+ value_name: str,
+ *,
+ minimum: int,
+ maximum: int,
+ ) -> int:
+ """Return an integer that is safe to embed in SQL grammar positions."""
+ if isinstance(value, bool) or not isinstance(value, int):
+ raise SQLSecurityError(f"Invalid {value_name}: must be an integer")
+ if not minimum <= value <= maximum:
+ raise SQLSecurityError(
+ f"Invalid {value_name}: must be between {minimum} and
{maximum}"
+ )
+ return value
+
+ @classmethod
+ def build_rule_predicate(
+ cls,
+ rule: Mapping[str, object],
+ ) -> tuple[str, tuple[object, ...]]:
+ """Build one parameterized business-rule predicate.
+
+ Raw SQL fragments are intentionally unsupported. Callers provide a
column,
+ an allowlisted operator, and a value (or values for ``IN``).
+ """
+ if "sql_condition" in rule:
+ raise SQLSecurityError(
+ "Raw sql_condition rules are not supported; use
column/operator/value"
+ )
+
+ column_name = rule.get("column")
+ if not isinstance(column_name, str):
+ raise SQLSecurityError("Business rule column must be a string")
+ column = cls.quote_identifier(column_name, "column name")
+ operator = str(rule.get("operator", "=")).strip().upper()
+
+ if operator in cls.NULL_RULE_OPERATORS:
+ return f"{column} {operator}", ()
+
+ if operator == "IN":
+ values = rule.get("values")
+ if (
+ not isinstance(values, Sequence)
+ or isinstance(values, str | bytes)
+ or not values
+ or len(values) > 100
+ ):
+ raise SQLSecurityError(
+ "IN rule values must be a non-empty sequence of at most
100 items"
+ )
+ placeholders = ", ".join("%s" for _ in values)
+ return f"{column} IN ({placeholders})", tuple(values)
+
+ if operator not in cls.RULE_OPERATORS:
+ allowed = sorted(
+ cls.RULE_OPERATORS | cls.NULL_RULE_OPERATORS | {"IN"}
+ )
+ raise SQLSecurityError(
+ f"Invalid business rule operator: {operator!r}; allowed:
{allowed}"
+ )
+
+ if "value" not in rule:
+ raise SQLSecurityError("Business rule value is required")
+ return f"{column} {operator} %s", (rule["value"],)
@staticmethod
def get_auth_context():
@@ -296,5 +377,7 @@ validate_identifier = SQLSecurityUtils.validate_identifier
quote_identifier = SQLSecurityUtils.quote_identifier
build_table_reference = SQLSecurityUtils.build_table_reference
build_column_reference = SQLSecurityUtils.build_column_reference
+build_rule_predicate = SQLSecurityUtils.build_rule_predicate
+validate_integer = SQLSecurityUtils.validate_integer
get_auth_context = SQLSecurityUtils.get_auth_context
set_auth_context = SQLSecurityUtils.set_auth_context
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index df9099a..2f703fd 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -446,8 +446,68 @@ async def test_real_doris_tool_regression_paths(
password=doris_sandbox.settings.password,
)
missing_table = f"{doris_sandbox.table}_missing"
+ with doris_sandbox.admin_connection.cursor() as cursor:
+ cursor.execute(
+ f"INSERT INTO {doris_sandbox.qualified_table} VALUES (1, %s)",
+ (doris_sandbox.marker,),
+ )
async with _transport_client(transport, environment) as client:
+ basic_info_result = await client.call_tool(
+ "get_table_basic_info",
+ {
+ "table_name": doris_sandbox.table,
+ "db_name": doris_sandbox.settings.database,
+ },
+ )
+ assert basic_info_result.is_error is False
+ assert isinstance(basic_info_result.structured_content, dict)
+ assert basic_info_result.structured_content["row_count"] == 1
+ assert basic_info_result.structured_content["column_count"] == 2
+
+ column_analysis_result = await client.call_tool(
+ "analyze_columns",
+ {
+ "table_name": doris_sandbox.table,
+ "columns": ["id", "marker"],
+ "analysis_types": ["completeness"],
+ "sample_size": 10,
+ "db_name": doris_sandbox.settings.database,
+ },
+ )
+ assert column_analysis_result.is_error is False
+ assert isinstance(column_analysis_result.structured_content, dict)
+ assert column_analysis_result.structured_content["columns_analyzed"]
== 2
+ assert set(
+ column_analysis_result.structured_content["completeness_analysis"]
+ ) == {"id", "marker"}
+
+ injected_identifier_result = await client.call_tool(
+ "analyze_columns",
+ {
+ "table_name": (
+ f"{doris_sandbox.table}; "
+ f"DROP TABLE {doris_sandbox.table}"
+ ),
+ "columns": ["id"],
+ "analysis_types": ["completeness"],
+ "sample_size": 10,
+ "db_name": doris_sandbox.settings.database,
+ },
+ )
+ assert injected_identifier_result.is_error is True
+ assert isinstance(injected_identifier_result.structured_content, dict)
+ assert "Invalid table name" in (
+ injected_identifier_result.structured_content["error"]
+ )
+
+ intact_result, intact_payload = await _exec_query(
+ client,
+ f"SELECT COUNT(*) AS row_count FROM
{doris_sandbox.qualified_table}",
+ )
+ assert intact_result.is_error is False
+ assert intact_payload["data"] == [{"row_count": 1}]
+
profile_result = await client.call_tool(
"get_sql_profile",
{
diff --git a/test/protocol/stdio_capability_server.py
b/test/protocol/stdio_capability_server.py
index ad588fc..14adf19 100644
--- a/test/protocol/stdio_capability_server.py
+++ b/test/protocol/stdio_capability_server.py
@@ -20,6 +20,7 @@ import asyncio
import json
import logging
import tempfile
+from contextlib import asynccontextmanager
from types import SimpleNamespace
from mcp.server.stdio import stdio_server
@@ -38,6 +39,7 @@ from doris_mcp_server.protocol import create_doris_mcp_server
from doris_mcp_server.tools.tools_manager import DorisToolsManager
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.security_analytics_tools import
SecurityAnalyticsTools
@@ -83,6 +85,10 @@ class ProfileConnectionManager:
async def get_connection(self, session_id: str) -> ProfileConnection:
return self.connection
+ @asynccontextmanager
+ async def get_connection_context(self, session_id: str):
+ yield self.connection
+
class ProfileAnalyzer(SQLAnalyzer):
async def _get_query_id_by_trace_id(self, trace_id: str) -> str:
@@ -188,6 +194,15 @@ class OneToolManager:
self.freshness_router.data_governance_tools = FreshnessGovernanceTools(
connection_manager
)
+ self.freshness_router.data_quality_tools = DataQualityTools(
+ connection_manager,
+ config=SimpleNamespace(
+ data_quality=SimpleNamespace(
+ enable_batch_analysis=True,
+ max_columns_per_batch=20,
+ )
+ ),
+ )
self.role_analyzer =
RoleSecurityAnalyticsTools(RoleConnectionManager())
async def list_tools(self) -> list[Tool]:
@@ -228,6 +243,22 @@ class OneToolManager:
description="Exercise Doris 4 role metadata compatibility.",
input_schema={"type": "object", "properties": {}},
),
+ Tool(
+ name="analyze_columns",
+ description="Exercise production SQL identifier validation.",
+ input_schema={
+ "type": "object",
+ "properties": {
+ "table_name": {"type": "string"},
+ "columns": {
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "db_name": {"type": "string"},
+ },
+ "required": ["table_name", "columns"],
+ },
+ ),
]
async def call_tool(self, name: str, arguments: dict) -> str:
@@ -257,6 +288,10 @@ class OneToolManager:
return json.dumps(
await self.role_analyzer.analyze_data_access_patterns()
)
+ if name == "analyze_columns":
+ return json.dumps(
+ await self.freshness_router._analyze_columns_tool(arguments)
+ )
return "{}"
diff --git a/test/protocol/test_mcp_v2_protocol.py
b/test/protocol/test_mcp_v2_protocol.py
index aa816e6..1bc3002 100644
--- a/test/protocol/test_mcp_v2_protocol.py
+++ b/test/protocol/test_mcp_v2_protocol.py
@@ -835,6 +835,53 @@ async def
test_http_doris4_role_metadata_uses_public_grants_command():
assert roles["operator"]["users"] == ["root"]
[email protected]
+async def test_http_rejects_injected_sql_identifier_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,
+ "analyze_columns",
+ {
+ "table_name": "orders; DROP TABLE orders",
+ "columns": ["id"],
+ "db_name": "hhm_dt_sim",
+ },
+ ),
+ headers=modern_tool_headers("analyze_columns"),
+ )
+ assert rejected.status_code == 200
+ assert rejected.json()["result"]["isError"] is True
+ assert "Invalid table name" in (
+ rejected.json()["result"]["structuredContent"]["error"]
+ )
+
+ 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")
@@ -886,6 +933,7 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
"get_sql_profile",
"monitor_data_freshness",
"analyze_data_access_patterns",
+ "analyze_columns",
]
secret = "stdio-secret-sec-016"
error_result = await capable.call_tool(
@@ -912,6 +960,7 @@ async def
test_stdio_validates_capabilities_versions_and_process_survival():
"get_sql_profile",
"monitor_data_freshness",
"analyze_data_access_patterns",
+ "analyze_columns",
]
legacy_error = await legacy.read_resource("doris://table/missing")
assert (
@@ -1047,6 +1096,33 @@ async def
test_stdio_doris4_role_metadata_uses_public_grants_command():
assert list(roles) == ["operator"]
[email protected]
+async def test_stdio_rejects_injected_sql_identifier_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(
+ "analyze_columns",
+ {
+ "table_name": "orders; DROP TABLE orders",
+ "columns": ["id"],
+ "db_name": "hhm_dt_sim",
+ },
+ )
+ assert rejected.is_error is True
+ assert "Invalid table name" in rejected.structured_content["error"]
+
+ 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_sql_sink_audit.py
b/test/security/test_sql_sink_audit.py
new file mode 100644
index 0000000..a1c7b1d
--- /dev/null
+++ b/test/security/test_sql_sink_audit.py
@@ -0,0 +1,276 @@
+# 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.
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from doris_mcp_server.tools.resources_manager import DorisResourcesManager
+from doris_mcp_server.utils.analysis_tools import TableAnalyzer
+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.performance_analytics_tools import (
+ PerformanceAnalyticsTools,
+)
+from doris_mcp_server.utils.schema_extractor import MetadataExtractor
+from doris_mcp_server.utils.security_analytics_tools import
SecurityAnalyticsTools
+from doris_mcp_server.utils.sql_security_utils import (
+ SQLSecurityError,
+ build_rule_predicate,
+ validate_integer,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+SOURCE_ROOT = REPOSITORY_ROOT / "doris_mcp_server"
+
+
+def test_b608_has_no_untriaged_dynamic_sql_findings():
+ bandit = shutil.which("bandit")
+ if bandit is None:
+ pytest.skip("install the project dev extra to run the Bandit SQL sink
gate")
+ completed = subprocess.run(
+ [
+ bandit,
+ "-q",
+ "-r",
+ str(SOURCE_ROOT),
+ "-t",
+ "B608",
+ "-f",
+ "json",
+ ],
+ cwd=REPOSITORY_ROOT,
+ capture_output=True,
+ check=False,
+ text=True,
+ )
+
+ report = json.loads(completed.stdout)
+ assert report["results"] == [], completed.stdout
+ assert completed.returncode == 0
+
+
+def test_every_b608_exception_carries_local_source_validation_sink_rationale():
+ exceptions = 0
+ for path in SOURCE_ROOT.rglob("*.py"):
+ lines = path.read_text(encoding="utf-8").splitlines()
+ previous_exception = -1
+ for index, line in enumerate(lines):
+ if "# nosec B608" not in line:
+ continue
+ exceptions += 1
+ rationale = " ".join(
+ lines[max(previous_exception + 1, index - 50) : index + 1]
+ )
+ assert "SQL sink audit:" in rationale, f"missing rationale:
{path}:{index + 1}"
+ assert (
+ "execute" in rationale or "executor" in rationale
+ ), f"missing sink: {path}:{index + 1}"
+ previous_exception = index
+
+ assert exceptions > 0
+
+
[email protected]("value", [True, "10", 1.5, None])
+def test_sql_integer_validation_rejects_non_integers(value):
+ with pytest.raises(SQLSecurityError):
+ validate_integer(value, "limit", minimum=0, maximum=100)
+
+
+def test_structured_business_rule_uses_bound_values():
+ attack_value = "active' OR 1=1 --"
+
+ predicate, params = build_rule_predicate(
+ {
+ "column": "status",
+ "operator": "=",
+ "value": attack_value,
+ }
+ )
+
+ assert predicate == "`status` = %s"
+ assert params == (attack_value,)
+ assert attack_value not in predicate
+
+
[email protected](
+ "rule",
+ [
+ {"sql_condition": "1=1); DROP TABLE users; --"},
+ {"sql_condition": ""},
+ {"operator": "=", "value": 1},
+ {"column": "status`; DROP TABLE users; --", "operator": "=", "value":
1},
+ {"column": "status", "operator": "= 1 OR 1=1", "value": 1},
+ {"column": "status", "operator": "IN", "values": []},
+ ],
+)
+def test_structured_business_rule_rejects_raw_or_invalid_fragments(rule):
+ with pytest.raises(SQLSecurityError):
+ build_rule_predicate(rule)
+
+
[email protected]
+async def test_business_rule_sink_binds_untrusted_value():
+ attack_value = "active' OR 1=1 --"
+ connection = SimpleNamespace(
+ execute=AsyncMock(
+ return_value=SimpleNamespace(
+ data=[{"total_count": 2, "pass_count": 1}]
+ )
+ )
+ )
+ tools = DataGovernanceTools(SimpleNamespace())
+
+ result = await tools._check_business_rule_compliance(
+ connection,
+ "`internal`.`analytics`.`orders`",
+ [
+ {
+ "rule_name": "active-only",
+ "column": "status",
+ "operator": "=",
+ "value": attack_value,
+ }
+ ],
+ 2,
+ )
+
+ assert result["active-only"]["pass_count"] == 1
+ sql = connection.execute.await_args.args[0]
+ assert attack_value not in sql
+ assert connection.execute.await_args.kwargs["params"] == (attack_value,)
+
+
[email protected]
+async def test_table_analyzer_rejects_identifier_injection_before_db_sink():
+ manager = SimpleNamespace(get_connection=AsyncMock())
+ analyzer = TableAnalyzer(manager)
+
+ result = await analyzer.analyze_column(
+ "orders`; DROP TABLE orders; --",
+ "amount",
+ )
+
+ assert result["success"] is False
+ manager.get_connection.assert_not_awaited()
+
+
[email protected]
+async def
test_batch_column_analysis_quotes_metadata_identifiers_and_stable_aliases():
+ connection = SimpleNamespace(
+ execute=AsyncMock(
+ return_value=SimpleNamespace(
+ data=[
+ {
+ "total_rows": 2,
+ "column_0_non_null": 1,
+ "column_0_distinct": 1,
+ }
+ ]
+ )
+ )
+ )
+ tools = object.__new__(DataQualityTools)
+
+ result = await tools._analyze_completeness_batch(
+ connection,
+ "`internal`.`analytics`.`orders`",
+ [{"column_name": "status"}],
+ )
+
+ sql = connection.execute.await_args.args[0]
+ assert "COUNT(`status`) as column_0_non_null" in sql
+ assert result["status"]["null_count"] == 1
+
+
+def test_resource_schema_filter_quotes_the_only_dynamic_identifier():
+ manager = DorisResourcesManager(SimpleNamespace())
+
+ condition, params = manager._schema_filter("table_schema", "analytics")
+
+ assert condition == "`table_schema` = %s"
+ assert params == ("analytics",)
+ with pytest.raises(SQLSecurityError):
+ manager._schema_filter("table_schema OR 1=1", "analytics")
+
+
[email protected]
+async def
test_schema_metadata_values_are_bound_before_connection_manager_sink():
+ manager = SimpleNamespace(
+ execute_query=AsyncMock(
+ return_value=SimpleNamespace(data=[{"TABLE_VISIBLE": 1}])
+ )
+ )
+ extractor = MetadataExtractor(
+ db_name="analytics",
+ catalog_name="internal",
+ connection_manager=manager,
+ )
+ extractor._is_doris_oauth_context = lambda auth_context=None: True
+ attack_table = "orders' OR 1=1 --"
+ attack_database = "analytics' OR 1=1 --"
+
+ await extractor._ensure_table_visible_for_doris_oauth_metadata(
+ attack_table,
+ attack_database,
+ "internal",
+ )
+
+ sql = manager.execute_query.await_args.args[1]
+ params = manager.execute_query.await_args.args[2]
+ assert attack_table not in sql
+ assert attack_database not in sql
+ assert params == (attack_database, attack_table)
+
+
[email protected]
+async def test_audit_analytics_bind_dates_thresholds_and_system_users():
+ connection = SimpleNamespace(
+ execute=AsyncMock(return_value=SimpleNamespace(data=[]))
+ )
+ performance = PerformanceAnalyticsTools(SimpleNamespace())
+ security = SecurityAnalyticsTools(SimpleNamespace())
+
+ await performance._get_slow_query_data(connection, 7, 1000)
+ performance_sql = connection.execute.await_args.args[0]
+ performance_params = connection.execute.await_args.kwargs["params"]
+ assert "WHERE `time` >= %s" in performance_sql
+ assert "`query_time` >= %s" in performance_sql
+ assert performance_params[1] == 1000
+
+ connection.execute.reset_mock()
+ await security._get_audit_log_data(
+ connection,
+ SimpleNamespace(),
+ SimpleNamespace(),
+ include_system_users=False,
+ )
+ security_sql = connection.execute.await_args.args[0]
+ security_params = connection.execute.await_args.kwargs["params"]
+ assert "NOT IN (%s, %s, %s, %s, %s)" in security_sql
+ assert security_params[2:] == (
+ "root",
+ "admin",
+ "system",
+ "doris",
+ "information_schema",
+ )
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
