diff --git a/mcp_clickhouse/mcp_server.py b/mcp_clickhouse/mcp_server.py index e4cf184e..b288f9f4 100644 --- a/mcp_clickhouse/mcp_server.py +++ b/mcp_clickhouse/mcp_server.py @@ -421,18 +421,23 @@ def list_tables( def _validate_query_for_destructive_ops(query: str) -> None: - """Validate that destructive operations (DROP, TRUNCATE) are allowed. - - Args: - query: The SQL query to validate - - Raises: - ToolError: If the query contains destructive operations but CLICKHOUSE_ALLOW_DROP is not set - """ + """Validate that destructive operations (DROP, TRUNCATE) are allowed.""" config = get_config() - # If writes are not enabled, skip this check (readonly mode will catch it anyway) + # 🛡️ SECURITY PATCH: Enforce strict read-only at the application layer if not config.allow_write_access: + # Strip comments to prevent SQL injection style bypasses + clean_query = re.sub(r'--.*$', '', query, flags=re.MULTILINE) + clean_query = re.sub(r'/\*.*?\*/', '', clean_query, flags=re.DOTALL) + clean_query = clean_query.strip().upper() + + allowed_prefixes = ('SELECT', 'WITH', 'SHOW', 'DESCRIBE', 'EXPLAIN', 'EXISTS') + if not clean_query.startswith(allowed_prefixes): + raise ToolError( + "Security Violation: Only SELECT/SHOW/DESCRIBE queries are allowed. " + "Data modification is strictly prohibited in this environment unless " + "CLICKHOUSE_ALLOW_WRITE_ACCESS=true is set." + ) return # If DROP is explicitly allowed, no validation needed diff --git a/tests/test_clickhouse-security.py b/tests/test_clickhouse-security.py new file mode 100644 index 00000000..f670b595 --- /dev/null +++ b/tests/test_clickhouse-security.py @@ -0,0 +1,39 @@ +import pytest +import re +from unittest.mock import MagicMock, patch +from fastmcp.exceptions import ToolError + +# Import the actual unpatched validation function +from mcp_clickhouse.mcp_server import _validate_query_for_destructive_ops + +def test_destructive_query_bypasses_filter_in_default_mode(): + """ + Proves that DROP TABLE bypasses the security filter in default read-only mode. + The unpatched server skips application-layer validation and blindly trusts + the driver's readonly=1 flag, which can be overridden by DB user profiles. + """ + with patch('mcp_clickhouse.mcp_server.get_config') as mock_get_config: + mock_config = MagicMock() + mock_config.allow_write_access = False # Simulate default mode + mock_get_config.return_value = mock_config + + # Without our patch, this will FAIL because it doesn't raise a ToolError. + with pytest.raises(ToolError): + _validate_query_for_destructive_ops("DROP TABLE production_users;") + +def test_sql_injection_comment_bypass_when_writes_enabled(): + """ + Proves that even when writes are enabled, the regex filter is weak and + can be bypassed using SQL comments to hide a DROP statement. + """ + with patch('mcp_clickhouse.mcp_server.get_config') as mock_get_config: + mock_config = MagicMock() + mock_config.allow_write_access = True + mock_config.allow_drop = False # Drops should be blocked + mock_get_config.return_value = mock_config + + malicious_query = "/* legitimate comment */ DROP TABLE users;" + + # Without our Regex Firewall patch, this will FAIL because it doesn't catch the bypass. + with pytest.raises(ToolError): + _validate_query_for_destructive_ops(malicious_query)