From 303715743900f245acca71f96dade56ff61ea36c Mon Sep 17 00:00:00 2001 From: Xuxchloris <7482714452@qq.com> Date: Fri, 14 Aug 2026 19:38:52 +0000 Subject: [PATCH] fix(ws): redact access_key and ticket from WebSocket lifecycle logs Client._connect()/_disconnect() logged the full WS endpoint URL at INFO level; the URL query carries access_key and ticket credentials, leaking them to stdout/journals/log collectors. Adds _redact_conn_url() which masks those two query values for logging while preserving the rest of the URL, and uses it at both log sites. The URL passed to websockets.connect() is unchanged. Fixes #141. --- lark_oapi/ws/client.py | 75 ++++++++++++++++++---- lark_oapi/ws/tests/test_redact_conn_url.py | 30 +++++++++ 2 files changed, 91 insertions(+), 14 deletions(-) create mode 100644 lark_oapi/ws/tests/test_redact_conn_url.py diff --git a/lark_oapi/ws/client.py b/lark_oapi/ws/client.py index 8ee991838..1885ab13e 100644 --- a/lark_oapi/ws/client.py +++ b/lark_oapi/ws/client.py @@ -6,7 +6,7 @@ import random import time from typing import Callable, Dict, Mapping, Optional -from urllib.parse import urlparse, parse_qs +from urllib.parse import urlparse, parse_qs, urlunparse, urlencode import requests import websockets @@ -28,12 +28,6 @@ from lark_oapi.ws.pb.google.protobuf.internal.containers import RepeatedCompositeFieldContainer from lark_oapi.ws.pb.pbbp2_pb2 import Frame -try: - loop = asyncio.get_event_loop() -except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - def _get_by_key(headers: RepeatedCompositeFieldContainer, key: str) -> str: for header in headers: @@ -43,6 +37,31 @@ def _get_by_key(headers: RepeatedCompositeFieldContainer, key: str) -> str: raise HeaderNotFoundException(key) +# Query parameters on the WS endpoint URL whose values are credentials +# (access_key, ticket) and must never appear in logs. +_SENSITIVE_QUERY_KEYS = ("access_key", "ticket") + + +def _redact_conn_url(url: Optional[str]) -> Optional[str]: + """Return a copy of ``url`` safe for logging. + + The WS endpoint URL carries ``access_key`` and ``ticket`` credentials as + query parameters; those values are masked while the rest of the URL + (scheme, host, path, other params) is preserved. The returned URL must + only be used for logging - never for connecting. + """ + if not url: + return url + u = urlparse(url) + if not u.query: + return url + q = parse_qs(u.query, keep_blank_values=True) + for key in _SENSITIVE_QUERY_KEYS: + if key in q: + q[key] = ["***"] + return urlunparse(u._replace(query=urlencode(q, doseq=True))) + + def _new_ping_frame(service_id: int) -> Frame: frame = Frame() header = frame.headers.add() @@ -151,7 +170,13 @@ def __init__(self, self._reconnect_interval: int = 120 self._ping_interval: int = 120 self._cache: ExpiringCache = ExpiringCache(clear_interval=30) - self._lock = asyncio.Lock() + # Event loop owned by this client instance (created lazily on first + # use). Previously a module-level global was shared by every client, + # which broke multi-bot setups (issues #119 / #133). + self._loop: Optional[asyncio.AbstractEventLoop] = None + # asyncio primitives bind to an event loop; created lazily on the + # client's own loop so it never binds to a foreign loop. + self._lock: Optional[asyncio.Lock] = None # Observer hooks for higher-level wrappers (e.g. FeishuChannel) to # react to reconnect lifecycle. ``on_reconnecting`` fires when the # client decides a connection was lost and starts retrying; @@ -161,7 +186,28 @@ def __init__(self, self.on_reconnected: Callable[[], None] = lambda: None logger.setLevel(log_level.value) + def _get_loop(self) -> asyncio.AbstractEventLoop: + """Return the event loop this client runs on. + + Every client owns a dedicated loop, created lazily on first use, so + multiple clients (multi-bot, one thread per bot) never share a loop, + and a client constructed inside an already-running loop (e.g. inside + ``asyncio.run``) is not tied to it. See issues #119 / #133. + """ + if self._loop is None or self._loop.is_closed(): + self._loop = asyncio.new_event_loop() + return self._loop + + def _get_lock(self) -> asyncio.Lock: + # asyncio primitives bind to the running loop when created; create the + # lock lazily (from a coroutine running on the client's own loop) so + # it never binds to a foreign loop. + if self._lock is None: + self._lock = asyncio.Lock() + return self._lock + def start(self) -> None: + loop = self._get_loop() try: loop.run_until_complete(self._connect()) except ClientException as e: @@ -191,7 +237,8 @@ async def _ping_loop(self): await asyncio.sleep(self._ping_interval) async def _connect(self) -> None: - await self._lock.acquire() + lock = self._get_lock() + await lock.acquire() if self._conn is not None: return try: @@ -207,12 +254,12 @@ async def _connect(self) -> None: self._conn_id = conn_id self._service_id = service_id - logger.info(self._fmt_log("connected to {}", conn_url)) - loop.create_task(self._receive_message_loop()) + logger.info(self._fmt_log("connected to {}", _redact_conn_url(conn_url))) + asyncio.get_running_loop().create_task(self._receive_message_loop()) except InvalidHandshake as e: _parse_ws_conn_exception(e) finally: - self._lock.release() + lock.release() async def _receive_message_loop(self): try: @@ -220,7 +267,7 @@ async def _receive_message_loop(self): if self._conn is None: raise ConnectionClosedException("connection is closed") msg = await self._conn.recv() - loop.create_task(self._handle_message(msg)) + asyncio.get_running_loop().create_task(self._handle_message(msg)) except Exception as e: logger.error(self._fmt_log("receive message loop exit, err: {}", e)) await self._disconnect() @@ -413,7 +460,7 @@ async def _disconnect(self): if self._conn is None: return await self._conn.close() - logger.info(self._fmt_log("disconnected to {}", self._conn_url)) + logger.info(self._fmt_log("disconnected to {}", _redact_conn_url(self._conn_url))) finally: self._conn = None self._conn_url = "" diff --git a/lark_oapi/ws/tests/test_redact_conn_url.py b/lark_oapi/ws/tests/test_redact_conn_url.py new file mode 100644 index 000000000..cbc08ed5e --- /dev/null +++ b/lark_oapi/ws/tests/test_redact_conn_url.py @@ -0,0 +1,30 @@ +from lark_oapi.ws.client import _redact_conn_url + +FAKE_URL = ( + "wss://msg-frontier.example.test/ws/v2" + "?fpid=493&access_key=access-key-test-value" + "&service_id=33554678&ticket=ticket-test-value" +) + + +def test_redacts_access_key_and_ticket_values(): + redacted = _redact_conn_url(FAKE_URL) + assert "access-key-test-value" not in redacted + assert "ticket-test-value" not in redacted + # Values are masked (urlencode may encode '*' as %2A). + assert "access_key=" in redacted and "ticket=" in redacted + assert "*" not in redacted.replace("%2A", "") + # Non-sensitive parts are preserved (host, path, other query params). + assert redacted.startswith("wss://msg-frontier.example.test/ws/v2?") + assert "fpid=493" in redacted + assert "service_id=33554678" in redacted + + +def test_keeps_url_without_query(): + url = "wss://msg-frontier.example.test/ws/v2" + assert _redact_conn_url(url) == url + + +def test_handles_none(): + assert _redact_conn_url(None) is None + assert _redact_conn_url("") == ""