runtime_bootstrap.py

import logging
import ssl
from typing import Any

import aiohttp.connector as aiohttp_connector

from astrbot.utils.http_ssl_common import build_ssl_context_with_certifi

logger = logging.getLogger(__name__)


def _try_patch_aiohttp_ssl_context(
    ssl_context: ssl.SSLContext,
    log_obj: Any | None = None,
) -> bool:
    log = log_obj or logger
    attr_name = "_SSL_CONTEXT_VERIFIED"

    if not hasattr(aiohttp_connector, attr_name):
        log.warning(
            "aiohttp connector does not expose _SSL_CONTEXT_VERIFIED; skipped patch.",
        )
        return False

    current_value = getattr(aiohttp_connector, attr_name, None)
    if current_value is not None and not isinstance(current_value, ssl.SSLContext):
        log.warning(
            "aiohttp connector exposes _SSL_CONTEXT_VERIFIED with unexpected type; skipped patch.",
        )
        return False

    setattr(aiohttp_connector, attr_name, ssl_context)
    log.info("Configured aiohttp verified SSL context with system+certifi trust chain.")
    return True


def configure_runtime_ca_bundle(log_obj: Any | None = None) -> bool:
    log = log_obj or logger

    try:
        log.info("Bootstrapping runtime CA bundle.")
        ssl_context = build_ssl_context_with_certifi(log_obj=log)
        return _try_patch_aiohttp_ssl_context(ssl_context, log_obj=log)
    except Exception as exc:
        log.error("Failed to configure runtime CA bundle for aiohttp: %r", exc)
        return False


def initialize_runtime_bootstrap(log_obj: Any | None = None) -> bool:
    return configure_runtime_ca_bundle(log_obj=log_obj)