fix(weixin): macOS SSL cert, QR data, and refresh rendering

- Use certifi CA bundle for aiohttp SSL in qr_login(), start(), and
  send_weixin_direct() to fix SSL verification failures against
  Tencent's iLink server on macOS (Homebrew OpenSSL lacks system certs)
- Fix QR code data: encode qrcode_img_content (full liteapp URL) instead
  of raw hex token — WeChat needs the full URL to resolve the scan
- Render ASCII QR on refresh so the user can re-scan without restarting
- Improve error message on QR render failure to show the actual exception

Tested on macOS (Apple Silicon, Homebrew Python 3.13)
This commit is contained in:
shenuu 2026-04-13 10:20:15 +08:00 committed by Teknium
parent e105b7ac93
commit 3a0ec1d935

View file

@ -98,6 +98,26 @@ MEDIA_VOICE = 4
_LIVE_ADAPTERS: Dict[str, Any] = {} _LIVE_ADAPTERS: Dict[str, Any] = {}
def _make_ssl_connector() -> Optional["aiohttp.TCPConnector"]:
"""Return a TCPConnector with a certifi CA bundle, or None if certifi is unavailable.
Tencent's iLink server (``ilinkai.weixin.qq.com``) is not verifiable against
some system CA stores (notably Homebrew's OpenSSL on macOS Apple Silicon).
When ``certifi`` is installed, use its Mozilla CA bundle to guarantee
verification. Otherwise fall back to aiohttp's default (which honors
``SSL_CERT_FILE`` env var via ``trust_env=True``).
"""
try:
import ssl
import certifi
except ImportError:
return None
if not AIOHTTP_AVAILABLE:
return None
ssl_ctx = ssl.create_default_context(cafile=certifi.where())
return aiohttp.TCPConnector(ssl=ssl_ctx)
ITEM_TEXT = 1 ITEM_TEXT = 1
ITEM_IMAGE = 2 ITEM_IMAGE = 2
ITEM_VOICE = 3 ITEM_VOICE = 3
@ -969,7 +989,7 @@ async def qr_login(
if not AIOHTTP_AVAILABLE: if not AIOHTTP_AVAILABLE:
raise RuntimeError("aiohttp is required for Weixin QR login") raise RuntimeError("aiohttp is required for Weixin QR login")
async with aiohttp.ClientSession(trust_env=True) as session: async with aiohttp.ClientSession(trust_env=True, connector=_make_ssl_connector()) as session:
try: try:
qr_resp = await _api_get( qr_resp = await _api_get(
session, session,
@ -987,6 +1007,10 @@ async def qr_login(
logger.error("weixin: QR response missing qrcode") logger.error("weixin: QR response missing qrcode")
return None return None
# qrcode_url is the full scannable liteapp URL; qrcode_value is just the hex token
# WeChat needs to scan the full URL, not the raw hex string
qr_scan_data = qrcode_url if qrcode_url else qrcode_value
print("\n请使用微信扫描以下二维码:") print("\n请使用微信扫描以下二维码:")
if qrcode_url: if qrcode_url:
print(qrcode_url) print(qrcode_url)
@ -994,11 +1018,11 @@ async def qr_login(
import qrcode import qrcode
qr = qrcode.QRCode() qr = qrcode.QRCode()
qr.add_data(qrcode_url or qrcode_value) qr.add_data(qr_scan_data)
qr.make(fit=True) qr.make(fit=True)
qr.print_ascii(invert=True) qr.print_ascii(invert=True)
except Exception: except Exception as _qr_exc:
print("(终端二维码渲染失败,请直接打开上面的二维码链接)") print(f"(终端二维码渲染失败: {_qr_exc},请直接打开上面的二维码链接)")
deadline = time.time() + timeout_seconds deadline = time.time() + timeout_seconds
current_base_url = ILINK_BASE_URL current_base_url = ILINK_BASE_URL
@ -1044,8 +1068,17 @@ async def qr_login(
) )
qrcode_value = str(qr_resp.get("qrcode") or "") qrcode_value = str(qr_resp.get("qrcode") or "")
qrcode_url = str(qr_resp.get("qrcode_img_content") or "") qrcode_url = str(qr_resp.get("qrcode_img_content") or "")
qr_scan_data = qrcode_url if qrcode_url else qrcode_value
if qrcode_url: if qrcode_url:
print(qrcode_url) print(qrcode_url)
try:
import qrcode as _qrcode
qr = _qrcode.QRCode()
qr.add_data(qr_scan_data)
qr.make(fit=True)
qr.print_ascii(invert=True)
except Exception:
pass
except Exception as exc: except Exception as exc:
logger.error("weixin: QR refresh failed: %s", exc) logger.error("weixin: QR refresh failed: %s", exc)
return None return None
@ -1169,8 +1202,8 @@ class WeixinAdapter(BasePlatformAdapter):
except Exception as exc: except Exception as exc:
logger.debug("[%s] Token lock unavailable (non-fatal): %s", self.name, exc) logger.debug("[%s] Token lock unavailable (non-fatal): %s", self.name, exc)
self._poll_session = aiohttp.ClientSession(trust_env=True) self._poll_session = aiohttp.ClientSession(trust_env=True, connector=_make_ssl_connector())
self._send_session = aiohttp.ClientSession(trust_env=True) self._send_session = aiohttp.ClientSession(trust_env=True, connector=_make_ssl_connector())
self._token_store.restore(self._account_id) self._token_store.restore(self._account_id)
self._poll_task = asyncio.create_task(self._poll_loop(), name="weixin-poll") self._poll_task = asyncio.create_task(self._poll_loop(), name="weixin-poll")
self._mark_connected() self._mark_connected()
@ -1964,7 +1997,7 @@ async def send_weixin_direct(
"context_token_used": bool(context_token), "context_token_used": bool(context_token),
} }
async with aiohttp.ClientSession(trust_env=True) as session: async with aiohttp.ClientSession(trust_env=True, connector=_make_ssl_connector()) as session:
adapter = WeixinAdapter( adapter = WeixinAdapter(
PlatformConfig( PlatformConfig(
enabled=True, enabled=True,