dbt MCP Server Logs Tool Arguments Including SQL Queries and Credentials in Plaintext Without Redaction When File Logging Is Enabled (GHSA-7xgw-6qf3-7w59) · Sieve
낮음취약점
dbt MCP Server Logs Tool Arguments Including SQL Queries and Credentials in Plaintext Without Redaction When File Logging Is Enabled
MCP client script — triggers real tool calls and verifies log file contents:
#!/usr/bin/env python3
# poc4_tool_args_logged.py
# Vulnerable code: src/dbt_mcp/mcp/server.py line 67, 77-79
# configure_file_logging(): src/dbt_mcp/telemetry/logging.py
import logging
from pathlib import Path
LOG_FILENAME = "dbt-mcp.log"
def configure_file_logging(log_level: int = logging.INFO) -> Path:
"""Reproduction of configure_file_logging() from telemetry/logging.py."""
module_path = Path(__file__).resolve().parent
home = Path.home().resolve()
for candidate in [module_path, *module_path.parents]:
if (candidate / ".git").exists() or (candidate / "pyproject.toml").exists() or candidate == home:
repo_root = candidate
break
log_path = repo_root / LOG_FILENAME
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
file_handler = logging.FileHandler(log_path, encoding="utf-8")
file_handler.setLevel(log_level)
file_handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s")
)
root_logger.addHandler(file_handler)
return log_path
log_path = configure_file_logging()
server_logger = logging.getLogger("dbt_mcp.mcp.server")
# Exact log statements from server.py line 67 and line 77-79
name = "show"
arguments = {"sql_query": "SELECT ssn, credit_card_number, salary FROM customers WHERE id = 42", "limit": 5}
server_logger.info(f"Calling tool: {name} with arguments: {arguments}")
name2 = "run"
arguments2 = {"node_selection": "sensitive_model", "vars": '{"db_password": "hunter2", "api_key": "sk-prod-abc123xyz"}', "is_full_refresh": False}
server_logger.info(f"Calling tool: {name2} with arguments: {arguments2}")
# Verify file contents
lines = log_path.read_text(encoding="utf-8").splitlines()
poc_lines = [l for l in lines if "dbt_mcp.mcp.server" in l]
print(f"[log file: {log_path}]")
for line in poc_lines:
print(f" {line}")
keywords = ["ssn", "credit_card_number", "salary", "db_password", "api_key"]
found = [kw for kw in keywords if any(kw in l for l in poc_lines)]
if found:
print(f"\n[CONFIRMED] Sensitive keywords in plaintext log: {found}")
print(f"[CONFIRMED] No redaction applied. File persists at {log_path}")
Expected log file entries:
2026-04-27 ... INFO [dbt_mcp.mcp.server] Calling tool: show with arguments:
{'sql_query': 'SELECT ssn, credit_card_number, salary FROM customers', 'limit': 5}
2026-04-27 ... INFO [dbt_mcp.mcp.server] Calling tool: run with arguments:
{'node_selection': 'sensitive_model',
'vars': '{"db_password":"hunter2","api_key":"sk-prod-abc123"}',
'is_full_refresh': False}
[CONFIRMED] Sensitive keywords in plaintext log: ['ssn', 'credit_card_number', 'salary', 'db_password', 'api_key']
[CONFIRMED] No redaction applied.
Impact
Directly proven by this PoC:
When DBT_MCP_SERVER_FILE_LOGGING=true, the full arguments dict of every tool call — including sql_query, vars, and node_selection — is written to dbt-mcp.log in plaintext on every invocation.
A tool call that raises an exception produces two log entries with the same sensitive content (INFO + ERROR double-logging).
The log file has no automatic rotation, expiry, or access restriction beyond filesystem permissions.
Combined with Advisory 3 (telemetry), a single show tool call containing PII produces one telemetry transmission to dbt Labs and one (or two, on failure) persistent log entries on disk.
Remediation
redact known-sensitive argument values before logging:
_LOG_REDACT = frozenset({"sql_query", "vars"})
def _safe_args(arguments: dict) -> dict:
return {k: "***redacted***" if k in _LOG_REDACT else v
for k, v in arguments.items()}
# server.py line 67:
logger.info(f"Calling tool: {name} with arguments: {_safe_args(arguments)}")
# server.py lines 77-79:
logger.error(
f"Error calling tool: {name} with arguments: {_safe_args(arguments)} "
f"in {end_time - start_time}ms: {e}"
)
log argument keys only:
logger.info(f"Calling tool: {name} with argument keys: {list(arguments.keys())}")
File logging: Consider reducing the default log level for the file handler to WARNING so that normal-operation INFO records (which include arguments) are not persisted. Sensitive content would only appear in file logs on error.