fix: harden frozen pip patch and unify TLS connector

astrbot/core/utils/http_ssl.py

@@ -0,0 +1,33 @@
+import logging
+import ssl
+
+import aiohttp
+import certifi
+
+logger = logging.getLogger("astrbot")
+
+_CERTIFI_WARNING_LOGGED = False
+
+
+def build_ssl_context_with_certifi() -> ssl.SSLContext:
+    """Build an SSL context from system trust store and add certifi CAs."""
+    global _CERTIFI_WARNING_LOGGED
+
+    ssl_context = ssl.create_default_context()
+
+    try:
+        ssl_context.load_verify_locations(cafile=certifi.where())
+    except Exception as exc:
+        if not _CERTIFI_WARNING_LOGGED:
+            logger.warning(
+                "Failed to load certifi CA bundle into SSL context; "
+                "falling back to system trust store only: %s",
+                exc,
+            )
+            _CERTIFI_WARNING_LOGGED = True
+
+    return ssl_context
+
+
+def build_tls_connector() -> aiohttp.TCPConnector:
+    return aiohttp.TCPConnector(ssl=build_ssl_context_with_certifi())

astrbot/core/utils/llm_metadata.py

@@ -1,10 +1,9 @@
-import ssl
 from typing import Literal, TypedDict
 
 import aiohttp
-import certifi
 
 from astrbot.core import logger
+from astrbot.core.utils.http_ssl import build_tls_connector
 
 
 class LLMModalities(TypedDict):
@@ -34,10 +33,8 @@ LLM_METADATAS: dict[str, LLMMetadata] = {}
 async def update_llm_metadata() -> None:
     url = "https://models.dev/api.json"
     try:
-        ssl_context = ssl.create_default_context(cafile=certifi.where())
-        connector = aiohttp.TCPConnector(ssl=ssl_context)
         async with aiohttp.ClientSession(
-            trust_env=True, connector=connector
+            trust_env=True, connector=build_tls_connector()
         ) as session:
             async with session.get(url) as response:
                 data = await response.json()

astrbot/core/utils/pip_installer.py

@@ -56,6 +56,21 @@ def _patch_distlib_finder_for_frozen_runtime() -> None:
     except Exception:
         return
 
+    finder_registry = getattr(distlib_resources, "_finder_registry", None)
+    register_finder = getattr(distlib_resources, "register_finder", None)
+    resource_finder = getattr(distlib_resources, "ResourceFinder", None)
+
+    if not isinstance(finder_registry, dict):
+        logger.warning(
+            "Skip patching distlib finder because _finder_registry is unavailable."
+        )
+        return
+    if not callable(register_finder) or resource_finder is None:
+        logger.warning(
+            "Skip patching distlib finder because register API is unavailable."
+        )
+        return
+
     for package_name in ("pip._vendor.distlib", "pip._vendor"):
         try:
             package = importlib.import_module(package_name)
@@ -69,10 +84,31 @@ def _patch_distlib_finder_for_frozen_runtime() -> None:
             continue
 
         loader_type = type(loader)
-        if loader_type in distlib_resources._finder_registry:
+        if loader_type in finder_registry:
+            continue
+
+        try:
+            register_finder(loader, resource_finder)
+        except Exception as exc:
+            logger.warning(
+                "Failed to patch pip distlib finder for loader %s (%s): %s",
+                loader_type.__name__,
+                package_name,
+                exc,
+            )
+            continue
+
+        finder_registry = getattr(
+            distlib_resources, "_finder_registry", finder_registry
+        )
+        if isinstance(finder_registry, dict) and loader_type not in finder_registry:
+            logger.warning(
+                "Distlib finder patch did not take effect for loader %s (%s).",
+                loader_type.__name__,
+                package_name,
+            )
             continue
 
-        distlib_resources.register_finder(loader, distlib_resources.ResourceFinder)
         logger.info(
             "Patched pip distlib finder for frozen loader: %s (%s)",
             loader_type.__name__,

astrbot/core/utils/t2i/network_strategy.py

@@ -1,12 +1,11 @@
 import asyncio
 import logging
 import random
-import ssl
 
 import aiohttp
-import certifi
 
 from astrbot.core.config import VERSION
+from astrbot.core.utils.http_ssl import build_tls_connector
 from astrbot.core.utils.io import download_image_by_url
 from astrbot.core.utils.t2i.template_manager import TemplateManager
 
@@ -39,11 +38,9 @@ class NetworkRenderStrategy(RenderStrategy):
     async def get_official_endpoints(self) -> None:
         """获取官方的 t2i 端点列表。"""
         try:
-            ssl_context = ssl.create_default_context(cafile=certifi.where())
-            connector = aiohttp.TCPConnector(ssl=ssl_context)
             async with aiohttp.ClientSession(
                 trust_env=True,
-                connector=connector,
+                connector=build_tls_connector(),
             ) as session:
                 async with session.get(
                     "https://api.soulter.top/astrbot/t2i-endpoints",
@@ -93,12 +90,10 @@ class NetworkRenderStrategy(RenderStrategy):
         for endpoint in endpoints:
             try:
                 if return_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,
+                            connector=build_tls_connector(),
                         ) as session,
                         session.post(
                             f"{endpoint}/generate",

main.py

@@ -8,12 +8,16 @@ from pathlib import Path
 
 def _configure_runtime_ca_bundle() -> None:
     try:
+        import ssl
+
+        import aiohttp.connector as aiohttp_connector
         import certifi
 
-        certifi_cafile = certifi.where()
-        if certifi_cafile and os.path.exists(certifi_cafile):
-            os.environ.setdefault("SSL_CERT_FILE", certifi_cafile)
-            os.environ.setdefault("REQUESTS_CA_BUNDLE", certifi_cafile)
+        ssl_context = ssl.create_default_context()
+        ssl_context.load_verify_locations(cafile=certifi.where())
+
+        if hasattr(aiohttp_connector, "_SSL_CONTEXT_VERIFIED"):
+            aiohttp_connector._SSL_CONTEXT_VERIFIED = ssl_context
     except Exception:
         return