diff --git a/src/langbot/pkg/api/http/controller/groups/platform/adapters.py b/src/langbot/pkg/api/http/controller/groups/platform/adapters.py index d740221f1..6b7f41d16 100644 --- a/src/langbot/pkg/api/http/controller/groups/platform/adapters.py +++ b/src/langbot/pkg/api/http/controller/groups/platform/adapters.py @@ -78,9 +78,42 @@ def _pop_owned_session( _MAX_ADAPTER_SESSIONS = 100 +# How often the credential-session sweeper enforces its TTL, independent of traffic. +_CREDENTIAL_SWEEP_INTERVAL_SECONDS = 60 _MAX_ADAPTER_SESSIONS_PER_WORKSPACE = 10 +def _scrub_credential_fields(session: dict) -> None: + """Overwrite credential fields before a session dict is dropped.""" + for key in ('app_secret', 'client_secret', 'secret'): + if key in session: + session[key] = None + + +def _sweep_credential_sessions(stores: 'list[tuple[dict, float]]') -> int: + """Evict sessions past their TTL and scrub the secrets they held. + + Returns the number of evicted sessions. Each entry is ``(store, ttl)``; a + session is evicted when ``now - created_at > ttl``. + """ + import time + + now = time.time() + evicted = 0 + for store, ttl in stores: + expired = [sid for sid, session in store.items() if now - session.get('created_at', 0) > ttl] + for sid in expired: + session = store.pop(sid, None) + if not session: + continue + task = session.get('task') + if task is not None and not task.done(): + task.cancel() + _scrub_credential_fields(session) + evicted += 1 + return evicted + + def _start_adapter_session_task( ap, coro, @@ -1231,3 +1264,44 @@ class AdaptersRouterGroup(group.RouterGroup): if session and session.get('task') and not session['task'].done(): session['task'].cancel() return self.success(data={}) + + # ---- credential-session TTL enforcement ---- + # + # Every store above holds a short-lived secret (app_secret, + # client_secret, bot secret) between the platform handshake and the + # caller persisting it. The per-route cleanups only run when another + # request arrives, so an abandoned flow would keep its secret in memory + # long past its validity window. This sweeper enforces the TTL on a + # timer instead, and scrubs the secret fields as it evicts. + + def _credential_stores() -> list[tuple[dict, float]]: + return [ + (_create_app_sessions, _SESSION_TTL), + (_dingtalk_sessions, _DINGTALK_SESSION_TTL), + (_wecombot_sessions, _WECOMBOT_SESSION_TTL), + (_qqofficial_sessions, _QQOFFICIAL_SESSION_TTL), + ] + + def _sweep_now() -> int: + return _sweep_credential_sessions(_credential_stores()) + + async def _credential_session_sweeper() -> None: + """Enforce the session TTL on a timer, so it does not depend on traffic.""" + import asyncio as _asyncio + + while True: + await _asyncio.sleep(_CREDENTIAL_SWEEP_INTERVAL_SECONDS) + try: + _sweep_now() + except Exception: # pragma: no cover - a sweep must never kill the app + _stdout_logger.exception('credential session sweep failed') + + self._sweep_credential_sessions = _sweep_now + # Some embeddings (notably test harnesses) provide a task manager without + # create_task; the per-route cleanups still enforce the TTL there. + create_task = getattr(self.ap.task_mgr, 'create_task', None) + if callable(create_task): + create_task( + _credential_session_sweeper(), + name='platform-credential-session-sweeper', + ) diff --git a/tests/unit_tests/api/test_credential_session_ttl.py b/tests/unit_tests/api/test_credential_session_ttl.py new file mode 100644 index 000000000..ed264fdcb --- /dev/null +++ b/tests/unit_tests/api/test_credential_session_ttl.py @@ -0,0 +1,102 @@ +"""Credential-session TTL enforcement for the platform adapter flows. + +Every adapter registration flow (Lark one-click app, DingTalk device flow, WeCom +bot, QQ Official bind) parks a short-lived secret in an in-memory session dict +between the platform handshake and the caller persisting it. The per-route +cleanups only run when another request arrives, so an abandoned flow would keep +its secret in memory past its validity window. These tests cover the sweeper that +enforces the TTL on a timer and scrubs the secret as it evicts. +""" + +import time + +from langbot.pkg.api.http.controller.groups.platform.adapters import ( + _sweep_credential_sessions, +) + + +def _session(age_seconds: float, **extra): + return {'status': 'pending', 'created_at': time.time() - age_seconds, **extra} + + +def test_expired_sessions_are_evicted(): + store = {'old': _session(1000)} + assert _sweep_credential_sessions([(store, 900)]) == 1 + assert store == {} + + +def test_fresh_sessions_survive(): + store = {'new': _session(10)} + assert _sweep_credential_sessions([(store, 900)]) == 0 + assert 'new' in store + + +def test_secret_is_scrubbed_on_eviction(): + """The evicted dict must not keep the credential readable.""" + session = _session(1000, app_secret='super-secret', client_secret='other', secret='third') + store = {'old': session} + _sweep_credential_sessions([(store, 900)]) + + assert store == {} + assert session['app_secret'] is None + assert session['client_secret'] is None + assert session['secret'] is None + + +def test_ttl_is_per_store(): + """A store with a shorter window expires while a longer one keeps its entry.""" + short, long = {'s': _session(400)}, {'l': _session(400)} + assert _sweep_credential_sessions([(short, 300), (long, 900)]) == 1 + assert short == {} + assert 'l' in long + + +def test_pending_task_is_cancelled_on_eviction(): + class FakeTask: + def __init__(self): + self.cancelled = False + + def done(self): + return False + + def cancel(self): + self.cancelled = True + + task = FakeTask() + store = {'old': _session(1000, task=task)} + _sweep_credential_sessions([(store, 900)]) + assert task.cancelled is True + + +def test_finished_task_is_not_cancelled(): + class DoneTask: + def __init__(self): + self.cancelled = False + + def done(self): + return True + + def cancel(self): # pragma: no cover - must not be called + self.cancelled = True + + task = DoneTask() + store = {'old': _session(1000, task=task)} + _sweep_credential_sessions([(store, 900)]) + assert task.cancelled is False + + +def test_session_without_created_at_is_evicted(): + """A malformed session must not become immortal.""" + store = {'broken': {'status': 'pending'}} + assert _sweep_credential_sessions([(store, 900)]) == 1 + assert store == {} + + +def test_multiple_expired_sessions_in_one_store(): + store = {f's{i}': _session(1000) for i in range(5)} + assert _sweep_credential_sessions([(store, 900)]) == 5 + assert store == {} + + +def test_empty_stores_are_harmless(): + assert _sweep_credential_sessions([({}, 900), ({}, 60)]) == 0