mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-11-03 22:24:42 +08:00
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GTnuEgR78cBk32bLNKar3z
This commit is contained in:
@@ -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))
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user