From edbc734d085451b470fde2670530b8e8536a3a41 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 2 Jul 2026 22:07:12 +0200 Subject: [PATCH] fix(security,sse): restore 72-byte password semantics, bound SSE downloader fetch - bcrypt 5.x raises on >72-byte passwords where passlib silently truncated; hash and verify now truncate to 72 bytes so pre-upgrade users with long (e.g. CJK) passwords can still log in instead of hitting HTTP 500 - The SSE downloader payload is fetched under a 3s asyncio.wait_for (task cancelled cleanly on timeout) and emits an explicit null on failure, so a dead qBittorrent can no longer freeze the entire event stream while the UI shows connected Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01GTnuEgR78cBk32bLNKar3z --- backend/src/module/api/events.py | 72 ++++++++++++++++---------- backend/src/module/security/jwt.py | 12 ++++- backend/src/test/test_api_events.py | 79 +++++++++++++++++++++++++++++ backend/src/test/test_auth.py | 45 ++++++++++++++++ 4 files changed, 179 insertions(+), 29 deletions(-) diff --git a/backend/src/module/api/events.py b/backend/src/module/api/events.py index 3f8963aa..777b59c3 100644 --- a/backend/src/module/api/events.py +++ b/backend/src/module/api/events.py @@ -30,6 +30,11 @@ _STATUS_EVERY = 3 _DOWNLOADER_EVERY = 5 _LOG_EVERY = 10 +# 下载器查询的超时上限（秒）。下载器不可达时 qB/aria2 的认证重试可能阻塞 +# 20-30s，而 SSE 是单连接串行推送——不设上限会把 status/log 事件一起卡住。 +# 取值需明显小于 downloader 的 5s 推送间隔。 +_DOWNLOADER_TIMEOUT_SECONDS = 3.0 + def _status_payload(ctx: AppContext) -> dict: """构造与 GET /program/status 相同结构的状态负载。""" @@ -40,11 +45,22 @@ def _status_payload(ctx: AppContext) -> dict: } +async def _fetch_torrents() -> list[dict]: + async with DownloadClient() as client: + return await client.get_torrent_info(category="Bangumi", status_filter=None) + + async def _downloader_payload() -> list[dict] | None: - """获取种子列表；下载器未配置或不可达时返回 None（跳过本次推送）。""" + """获取种子列表；下载器未配置、不可达或超时时返回 None。 + + asyncio.wait_for 超时会取消内部任务并等待其退出，慢查询不会泄漏； + None 会以 null 推送给前端，作为显式的"下载器不可用"信号。 + """ try: - async with DownloadClient() as client: - return await client.get_torrent_info(category="Bangumi", status_filter=None) + return await asyncio.wait_for(_fetch_torrents(), _DOWNLOADER_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + logger.debug("SSE: downloader status fetch timed out") + return None except Exception: logger.debug("SSE: downloader status unavailable", exc_info=True) return None @@ -58,30 +74,32 @@ async def _log_payload() -> str | None: return data.decode("utf-8", errors="replace") +async def _event_generator(request: Request, ctx: AppContext): + """按各自节拍推送 status/downloader/log 事件的核心循环。""" + tick = 0 + while True: + if await request.is_disconnected(): + break + + if tick % _STATUS_EVERY == 0: + yield {"event": "status", "data": json.dumps(_status_payload(ctx))} + + if tick % _DOWNLOADER_EVERY == 0: + # 不可用时推送 null 而非跳过，前端据此感知下载器降级状态 + # （store/downloader.ts 对 null 保留旧数据，不会崩溃）。 + torrents = await _downloader_payload() + yield {"event": "downloader", "data": json.dumps(torrents)} + + if tick % _LOG_EVERY == 0: + log_text = await _log_payload() + if log_text is not None: + yield {"event": "log", "data": log_text} + + await asyncio.sleep(_TICK_SECONDS) + tick += _TICK_SECONDS + + @router.get("/stream", dependencies=[Depends(get_current_user)]) async def event_stream(request: Request, ctx: AppContext = Depends(get_context)): """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件。""" - - async def generator(): - tick = 0 - while True: - if await request.is_disconnected(): - break - - if tick % _STATUS_EVERY == 0: - yield {"event": "status", "data": json.dumps(_status_payload(ctx))} - - if tick % _DOWNLOADER_EVERY == 0: - torrents = await _downloader_payload() - if torrents is not None: - yield {"event": "downloader", "data": json.dumps(torrents)} - - if tick % _LOG_EVERY == 0: - log_text = await _log_payload() - if log_text is not None: - yield {"event": "log", "data": log_text} - - await asyncio.sleep(_TICK_SECONDS) - tick += _TICK_SECONDS - - return EventSourceResponse(generator()) + return EventSourceResponse(_event_generator(request, ctx)) diff --git a/backend/src/module/security/jwt.py b/backend/src/module/security/jwt.py index 020d65d0..ecd77e91 100644 --- a/backend/src/module/security/jwt.py +++ b/backend/src/module/security/jwt.py @@ -58,11 +58,19 @@ def verify_token(token: str | None): # 密码加密&验证 +def _truncate_password(password: str) -> bytes: + # bcrypt 只使用前 72 字节；passlib 时代会静默截断，bcrypt 5.x 则对超长输入 + # 抛 ValueError。这里显式截断以保持 passlib 兼容语义：升级前用超长密码 + # （UTF-8 下中文很容易超过 72 字节）注册的用户仍可用原密码登录，设置 + # 超长新密码也不会报错。 + return password.encode("utf-8")[:72] + + def verify_password(plain_password: str, hashed_password: str) -> bool: return bcrypt.checkpw( - plain_password.encode("utf-8"), hashed_password.encode("utf-8") + _truncate_password(plain_password), hashed_password.encode("utf-8") ) def get_password_hash(password: str) -> str: - return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") + return bcrypt.hashpw(_truncate_password(password), bcrypt.gensalt()).decode("utf-8") diff --git a/backend/src/test/test_api_events.py b/backend/src/test/test_api_events.py index 9c9cf0ba..261175d8 100644 --- a/backend/src/test/test_api_events.py +++ b/backend/src/test/test_api_events.py @@ -1,5 +1,8 @@ """Tests for the events SSE endpoint and its payload builders.""" +import asyncio +import json +import time from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -102,6 +105,82 @@ class TestDownloaderPayload: assert result is None +def _make_hung_client(cancelled: asyncio.Event) -> AsyncMock: + """Build a DownloadClient mock whose torrent fetch hangs until cancelled, + setting `cancelled` when the hung task is actually torn down.""" + + async def hung_fetch(*args, **kwargs): + try: + await asyncio.sleep(30) + except asyncio.CancelledError: + cancelled.set() + raise + return [] + + mock_client = AsyncMock() + mock_client.get_torrent_info = hung_fetch + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + return mock_client + + +class TestDownloaderPayloadTimeout: + async def test_downloader_payload_hung_fetch_returns_none_and_cancels_task(self): + """A downloader fetch exceeding the timeout returns None promptly and + the hung task is cancelled rather than leaked.""" + cancelled = asyncio.Event() + mock_client = _make_hung_client(cancelled) + + with ( + patch("module.api.events.DownloadClient", return_value=mock_client), + patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05), + ): + start = time.monotonic() + result = await _downloader_payload() + elapsed = time.monotonic() - start + + assert result is None + assert elapsed < 1.0 + assert cancelled.is_set() + + +class TestEventGeneratorNonBlocking: + async def test_event_generator_hung_downloader_emits_status_and_null_downloader( + self, tmp_path + ): + """With an unreachable downloader, the stream still yields the status + event promptly and emits an explicit null downloader payload instead + of stalling the whole connection.""" + from module.api.events import _event_generator + + request = AsyncMock() + request.is_disconnected = AsyncMock(return_value=False) + ctx = MagicMock() + ctx.is_running = True + ctx.first_run = False + + cancelled = asyncio.Event() + mock_client = _make_hung_client(cancelled) + + with ( + patch("module.api.events.DownloadClient", return_value=mock_client), + patch("module.api.events._DOWNLOADER_TIMEOUT_SECONDS", 0.05), + patch("module.api.events.LOG_PATH", tmp_path / "missing.log"), + ): + gen = _event_generator(request, ctx) + try: + # 第一个 tick 内 status 与 downloader 事件都应在超时上限内到达 + status_event = await asyncio.wait_for(gen.__anext__(), timeout=1.0) + downloader_event = await asyncio.wait_for(gen.__anext__(), timeout=1.0) + finally: + await gen.aclose() + + assert status_event["event"] == "status" + assert json.loads(status_event["data"])["status"] is True + assert downloader_event["event"] == "downloader" + assert json.loads(downloader_event["data"]) is None + + class TestLogPayload: async def test_log_payload_missing_file_returns_none(self, tmp_path): """_log_payload returns None when the log file doesn't exist yet.""" diff --git a/backend/src/test/test_auth.py b/backend/src/test/test_auth.py index 9c0c06f4..be97ea34 100644 --- a/backend/src/test/test_auth.py +++ b/backend/src/test/test_auth.py @@ -144,6 +144,51 @@ class TestPasswordHashing: assert verify_password(password, hash2) is True +# --------------------------------------------------------------------------- +# Long (>72-byte) passwords — passlib silently truncated at 72 bytes, and +# bcrypt 5.x raises ValueError instead. The wrappers must keep passlib's +# truncation semantics so pre-upgrade users with long passwords can still +# log in (and setting a long password doesn't 500). +# --------------------------------------------------------------------------- + + +class TestLongPasswordTruncation: + # 30 CJK chars * 3 bytes (UTF-8) = 90 bytes > 72-byte bcrypt limit + LONG_CJK_PASSWORD = "密码超过七十二字节" * 4 + + def test_get_password_hash_over_72_byte_cjk_password_hashes_and_verifies(self): + """A >72-byte multibyte password hashes without error and verifies.""" + password = self.LONG_CJK_PASSWORD + assert len(password.encode("utf-8")) > 72 + hashed = get_password_hash(password) + assert verify_password(password, hashed) is True + + def test_verify_password_pre_upgrade_truncated_hash_full_password_verifies(self): + """A passlib-era hash (created from the first 72 bytes) still verifies + against the full long password after the bcrypt migration.""" + import bcrypt + + password = self.LONG_CJK_PASSWORD + legacy_hash = bcrypt.hashpw( + password.encode("utf-8")[:72], bcrypt.gensalt() + ).decode("utf-8") + assert verify_password(password, legacy_hash) is True + + def test_verify_password_wrong_long_password_fails(self): + """A long password differing within its first 72 bytes still fails.""" + hashed = get_password_hash(self.LONG_CJK_PASSWORD) + wrong = "错误密码完全不同的内容" * 4 + assert len(wrong.encode("utf-8")) > 72 + assert verify_password(wrong, hashed) is False + + def test_verify_password_over_72_byte_ascii_password_roundtrip(self): + """A >72-byte pure-ASCII password also hashes and verifies.""" + password = "a" * 100 + hashed = get_password_hash(password) + assert verify_password(password, hashed) is True + assert verify_password("b" * 100, hashed) is False + + # --------------------------------------------------------------------------- # API Auth Flow (get_current_user) # ---------------------------------------------------------------------------