From 2165475dd02bc8b9627f99fa6e19606c478b5475 Mon Sep 17 00:00:00 2001 From: freeman Date: Fri, 9 Oct 2026 11:53:46 +0800 Subject: [PATCH] fix(security): throttle password login after repeated failures The access-token endpoint accepted unlimited password attempts; the only existing rate limiters are single-instance helpers for outbound site requests. Add app/application/security/throttle.py, an in-memory LoginThrottle keyed by (client address, normalized username) that locks for 30s after five consecutive failures and doubles the lockout up to 15 minutes. The endpoint answers 429 with Retry-After while locked, counts only real credential failures (an MFA challenge after a correct password does not count), and clears the record on success. X-Forwarded-For is deliberately not trusted because there is no trusted-proxy setting; behind the bundled nginx all requests share one address, so the key degrades to per-username throttling, which is the dimension brute-force protection needs. Sync the dependency snapshot and architecture metrics for the new module. Co-Authored-By: Claude Fable 5.1 --- app/api/endpoints/login.py | 23 ++- app/application/security/throttle.py | 126 ++++++++++++++++ app/locales/en-US.json | 1 + app/locales/zh-TW.json | 1 + docs/architecture-overview.md | 6 +- docs/refactor/optimization-checklist.md | 2 +- docs/rules/11-quality-and-security.md | 4 + .../architecture/dependency-baseline.json | 8 +- tests/test_login_mfa_methods.py | 137 ++++++++++++++++++ tests/test_login_throttle.py | 132 +++++++++++++++++ 10 files changed, 431 insertions(+), 9 deletions(-) create mode 100644 app/application/security/throttle.py create mode 100644 tests/test_login_throttle.py diff --git a/app/api/endpoints/login.py b/app/api/endpoints/login.py index 7bd1928e4..0aaabf572 100644 --- a/app/api/endpoints/login.py +++ b/app/api/endpoints/login.py @@ -17,6 +17,7 @@ from app.api.response import ( ) from app.application.configuration import ApiRuntimeConfig, get_runtime_settings from app.application.image import WallpaperHelper +from app.application.security.throttle import build_login_throttle_key, get_login_throttle from app.application.security.token import PasswordTooLongError, create_access_token, get_password_hash from app.application.security.user import UserNameConflictError, UserService from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module @@ -148,7 +149,11 @@ async def initialize_instance( 401: { "model": _SchemaResponse[_SchemaMfaChallenge], "description": "需要二次验证或认证失败", - } + }, + 429: { + "model": _SchemaResponse[None], + "description": "连续登录失败过多，响应头 Retry-After 给出需等待的秒数", + }, }, openapi_extra={RAW_RESPONSE_OPENAPI_KEY: True}, ) @@ -163,12 +168,24 @@ def login_access_token( 获取认证Token """ runtime_config = resolve_api_runtime_config(runtime_config) + # 连续失败后按指数退避拒绝尝试，先于任何凭据校验，避免锁定期内继续消耗密码哈希 + throttle = get_login_throttle() + throttle_key = build_login_throttle_key( + request.client.host if request.client else None, form_data.username + ) + retry_after = throttle.retry_after(throttle_key) + if retry_after: + raise HTTPException( + status_code=429, + detail="登录尝试过于频繁，请稍后重试", + headers={"Retry-After": str(retry_after)}, + ) success, user_or_message = UserChain().user_authenticate( username=form_data.username, password=form_data.password, mfa_code=otp_password ) if not success: - # 只有密码已经验证通过时才返回 MFA 方法，避免泄露账号安全配置。 + # 只有密码已经验证通过时才返回 MFA 方法，避免泄露账号安全配置；密码正确不计失败。 if isinstance(user_or_message, MfaRequired): challenge = _SchemaResponse[_SchemaMfaChallenge]( success=False, @@ -182,8 +199,10 @@ def login_access_token( content=challenge.model_dump(mode="json"), headers={"X-MFA-Required": "true"}, ) + throttle.record_failure(throttle_key) raise HTTPException(status_code=401, detail="用户名或密码错误") + throttle.record_success(throttle_key) # 用户等级 level = SitesHelper().auth_level access_token = create_access_token( diff --git a/app/application/security/throttle.py b/app/application/security/throttle.py new file mode 100644 index 000000000..cc7c46823 --- /dev/null +++ b/app/application/security/throttle.py @@ -0,0 +1,126 @@ +"""密码登录的连续失败计数与指数退避锁定策略。""" + +import math +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import Optional + +# 连续失败达到该次数后开始锁定 +LOGIN_MAX_FAILURES = 5 +# 首次锁定时长，之后每再失败一次翻倍 +LOGIN_BASE_LOCK_SECONDS = 30 +# 单次锁定上限 +LOGIN_MAX_LOCK_SECONDS = 15 * 60 +# 记录在最后一次失败后保留多久 +LOGIN_THROTTLE_TTL_SECONDS = 60 * 60 +# 内存记录上限，超出后按最近更新时间淘汰最旧 +LOGIN_THROTTLE_MAX_ITEMS = 4096 +# 指数上限，避免 2 ** n 无意义增长 +_MAX_BACKOFF_EXPONENT = 20 + +LoginThrottleKey = tuple[str, str] + + +def build_login_throttle_key(client_host: Optional[str], username: Optional[str]) -> LoginThrottleKey: + """ + 构造登录限流键：(客户端地址, 归一化用户名)。 + + 只使用直连客户端地址，不解析 X-Forwarded-For；没有可信代理配置时盲信该头会让 + 攻击者换头绕过。同机反代场景下所有请求同地址，键退化为按用户名限流，这正是 + 暴力破解防护需要的维度；直接暴露服务时复合键避免单一地址扫号锁死所有用户。 + """ + return str(client_host or "").strip(), str(username or "").strip().lower() + + +@dataclass +class _Entry: + """单个键的失败计数与锁定状态。""" + + failures: int = 0 + locked_until: float = 0.0 + updated_at: float = 0.0 + + +class LoginThrottle: + """ + 按 (客户端地址, 用户名) 记录连续失败并施加指数锁定的进程内限流器。 + + 连续失败达到 ``max_failures`` 后锁定 ``base_lock_seconds``，之后每再失败一次 + 锁定时长翻倍直到 ``max_lock_seconds``；登录成功清除记录。状态只在当前进程 + 内存中，多 worker 进程之间不共享。 + """ + + def __init__( + self, + *, + max_failures: int = LOGIN_MAX_FAILURES, + base_lock_seconds: int = LOGIN_BASE_LOCK_SECONDS, + max_lock_seconds: int = LOGIN_MAX_LOCK_SECONDS, + ttl_seconds: int = LOGIN_THROTTLE_TTL_SECONDS, + max_items: int = LOGIN_THROTTLE_MAX_ITEMS, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._max_failures = max(1, int(max_failures)) + self._base_lock_seconds = max(1, int(base_lock_seconds)) + self._max_lock_seconds = max(self._base_lock_seconds, int(max_lock_seconds)) + self._ttl_seconds = max(1, int(ttl_seconds)) + self._max_items = max(1, int(max_items)) + self._clock = clock + self._entries: dict[LoginThrottleKey, _Entry] = {} + self._lock = threading.RLock() + + def retry_after(self, key: LoginThrottleKey) -> int: + """返回该键仍需等待的秒数（向上取整），0 表示允许尝试。""" + now = self._clock() + with self._lock: + entry = self._entries.get(key) + if entry is None: + return 0 + remaining = entry.locked_until - now + return math.ceil(remaining) if remaining > 0 else 0 + + def record_failure(self, key: LoginThrottleKey) -> int: + """记录一次失败，返回本次施加的锁定秒数，0 表示尚未达到锁定阈值。""" + now = self._clock() + with self._lock: + entry = self._entries.setdefault(key, _Entry()) + entry.failures += 1 + entry.updated_at = now + lock_seconds = 0 + if entry.failures >= self._max_failures: + exponent = min(entry.failures - self._max_failures, _MAX_BACKOFF_EXPONENT) + lock_seconds = min(self._base_lock_seconds * (2 ** exponent), self._max_lock_seconds) + entry.locked_until = now + lock_seconds + self._cleanup(now) + return lock_seconds + + def record_success(self, key: LoginThrottleKey) -> None: + """登录成功后清除该键的失败记录。""" + with self._lock: + self._entries.pop(key, None) + + def _cleanup(self, now: float) -> None: + """清理过期记录，并在超出容量时淘汰最久未更新的记录。""" + expired = [ + key + for key, entry in self._entries.items() + if entry.locked_until <= now and now - entry.updated_at > self._ttl_seconds + ] + for key in expired: + self._entries.pop(key, None) + overflow = len(self._entries) - self._max_items + if overflow <= 0: + return + ordered = sorted(self._entries.items(), key=lambda item: item[1].updated_at) + for key, _ in ordered[:overflow]: + self._entries.pop(key, None) + + +_default_throttle = LoginThrottle() + + +def get_login_throttle() -> LoginThrottle: + """返回进程内共享的登录限流器。""" + return _default_throttle diff --git a/app/locales/en-US.json b/app/locales/en-US.json index ff50f4fcc..7f9a57e52 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -465,6 +465,7 @@ "用户权限不足": "Insufficient user permissions", "用户名或密码错误": "Incorrect username or password", "需要二次验证": "Two-step verification is required", + "登录尝试过于频繁，请稍后重试": "Too many login attempts, please try again later", "图片读取出错": "Failed to read image", "授权失败": "Authorization failed", "报文内容为空": "Request payload is empty", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 17adf6d44..3490950e5 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -457,6 +457,7 @@ "用户权限不足": "使用者權限不足", "用户名或密码错误": "使用者名稱或密碼錯誤", "需要二次验证": "需要二次驗證", + "登录尝试过于频繁，请稍后重试": "登入嘗試過於頻繁，請稍後重試", "图片读取出错": "圖片讀取出錯", "授权失败": "授權失敗", "报文内容为空": "報文內容為空", diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 2aee5cf87..2a1c7186a 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -762,8 +762,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 1113 | -| 内部导入边 | 9,415 | +| Python 模块 | 1114 | +| 内部导入边 | 9,416 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 49（债务已清零，49 条精确 containment） | @@ -773,7 +773,7 @@ flowchart LR | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | -整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,415，音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 +整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,416，音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 25f11996e..8a7870a0c 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -98,7 +98,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1113 / 9,415 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 | +| 宿主 Python 模块 / 内部依赖边 | 1114 / 9,416 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/rules/11-quality-and-security.md b/docs/rules/11-quality-and-security.md index 511a71cad..8270bb9e2 100644 --- a/docs/rules/11-quality-and-security.md +++ b/docs/rules/11-quality-and-security.md @@ -84,6 +84,10 @@ All REST and MCP API endpoints require authentication. The project supports two The `API_TOKEN` value in `settings` is the source of truth. It is set at initialization and never exposed in logs or API responses. +### Login Throttling + +Password login (`/api/v1/login/access-token`) is protected by `app/application/security/throttle.py`: consecutive failures keyed by `(client address, normalized username)` trigger exponential lockouts (5 failures → 30s, doubling up to 15 minutes), and the endpoint answers `429` with a `Retry-After` header while locked. A correct password that still needs MFA does not count as a failure; a successful login clears the record. The state lives in process memory only, and `X-Forwarded-For` is deliberately not trusted because there is no trusted-proxy setting. + ### Endpoint Authorization - API-token authenticated integration endpoints are administrator-level surfaces unless a specific endpoint documents a narrower contract. diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 2eda43eff..a48251d5e 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -941,8 +941,8 @@ "runtime_only": true } }, - "edge_count": 9415, - "edge_sha256": "cf827acc39adbf6e1080ce8c09922c286c03423990dd331d90df30d9b59097e4", + "edge_count": 9416, + "edge_sha256": "b1ea9dffbfa52f3cf110294c10113b24f592aca38a97d4025e3fa99f9bee7005", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -2555,6 +2555,7 @@ "app.api.endpoints.login -> app.application.configuration", "app.api.endpoints.login -> app.application.image", "app.api.endpoints.login -> app.application.security", + "app.api.endpoints.login -> app.application.security.throttle", "app.api.endpoints.login -> app.application.security.token", "app.api.endpoints.login -> app.application.security.user", "app.api.endpoints.login -> app.application.site", @@ -10360,7 +10361,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 1113, + "module_count": 1114, "modules": [ "app", "app.adapters", @@ -10738,6 +10739,7 @@ "app.application.security.otp", "app.application.security.passkey", "app.application.security.secrets", + "app.application.security.throttle", "app.application.security.token", "app.application.security.twofactor", "app.application.security.url", diff --git a/tests/test_login_mfa_methods.py b/tests/test_login_mfa_methods.py index 29e517187..07b4c38c8 100644 --- a/tests/test_login_mfa_methods.py +++ b/tests/test_login_mfa_methods.py @@ -170,3 +170,140 @@ def test_passkey_registration_start_returns_object_options(monkeypatch): assert response.success is True assert payload.options.root["challenge"] == "register-challenge" assert payload.transaction_token == "registration-transaction" + + +class _FakeThrottle: + """记录调用并按预设返回等待秒数的登录限流器替身。""" + + def __init__(self, retry_after: int = 0) -> None: + self._retry_after = retry_after + self.failures: list[tuple[str, str]] = [] + self.successes: list[tuple[str, str]] = [] + + def retry_after(self, key): + """返回预设等待秒数。""" + return self._retry_after + + def record_failure(self, key): + """记录失败键。""" + self.failures.append(key) + return 0 + + def record_success(self, key): + """记录成功键。""" + self.successes.append(key) + + +def test_login_rejects_with_429_while_throttled(monkeypatch): + """锁定期内直接返回 429 与 Retry-After，不触发凭据校验。""" + throttle = _FakeThrottle(retry_after=42) + monkeypatch.setattr(login_endpoint, "get_login_throttle", lambda: throttle) + calls = [] + + class FakeUserChain: + """记录是否被调用。""" + + def user_authenticate(self, **kwargs): + """不应被调用。""" + calls.append(kwargs) + return False, "unexpected" + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + + with pytest.raises(HTTPException) as exc_info: + login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=_form(), + ) + + assert exc_info.value.status_code == 429 + assert exc_info.value.headers == {"Retry-After": "42"} + assert calls == [] + assert throttle.failures == [] + + +def test_login_failure_records_throttle_key(monkeypatch): + """密码错误按 (客户端地址, 小写用户名) 记一次失败。""" + throttle = _FakeThrottle() + monkeypatch.setattr(login_endpoint, "get_login_throttle", lambda: throttle) + + class FakeUserChain: + """返回普通认证失败。""" + + def user_authenticate(self, username, password, mfa_code=None): + """模拟错误密码。""" + return False, "用户名、密码或验证码错误" + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + + with pytest.raises(HTTPException) as exc_info: + login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=SimpleNamespace(username=" User ", password="password"), + ) + + assert exc_info.value.status_code == 401 + assert throttle.failures == [("testclient", "user")] + assert throttle.successes == [] + + +def test_login_mfa_challenge_does_not_count_as_failure(monkeypatch): + """密码正确但需二次验证时不计入失败。""" + throttle = _FakeThrottle() + monkeypatch.setattr(login_endpoint, "get_login_throttle", lambda: throttle) + + class FakeUserChain: + """返回已通过密码校验的 MFA 要求。""" + + def user_authenticate(self, username, password, mfa_code=None): + """模拟账号启用了 OTP。""" + return False, MfaRequired(methods=("otp",)) + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + + response = login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=_form(), + ) + + assert response.status_code == 401 + assert throttle.failures == [] + assert throttle.successes == [] + + +def test_login_success_clears_throttle(monkeypatch): + """登录成功后清除该键的失败记录。""" + throttle = _FakeThrottle() + monkeypatch.setattr(login_endpoint, "get_login_throttle", lambda: throttle) + user = SimpleNamespace( + id=1, name="user", is_superuser=True, avatar=None, permissions={} + ) + + class FakeUserChain: + """返回认证成功的用户。""" + + def user_authenticate(self, username, password, mfa_code=None): + """模拟认证成功。""" + return True, user + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + monkeypatch.setattr( + login_endpoint, "SitesHelper", lambda: SimpleNamespace(auth_level=2) + ) + monkeypatch.setattr(login_endpoint, "create_access_token", lambda **_: "token") + monkeypatch.setattr( + login_endpoint, "set_or_refresh_resource_token_cookie", lambda *_args, **_kwargs: None + ) + + token = login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=_form(), + ) + + assert token.access_token == "token" + assert throttle.successes == [("testclient", "user")] + assert throttle.failures == [] diff --git a/tests/test_login_throttle.py b/tests/test_login_throttle.py new file mode 100644 index 000000000..6f310a7ff --- /dev/null +++ b/tests/test_login_throttle.py @@ -0,0 +1,132 @@ +"""登录失败指数退避限流器的策略测试。""" + +from app.application.security.throttle import LoginThrottle, build_login_throttle_key, get_login_throttle + + +class _Clock: + """可手动推进的单调时钟。""" + + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + """推进时钟。""" + self.now += seconds + + +def _throttle(clock: _Clock, **overrides) -> LoginThrottle: + """构造使用假时钟的限流器。""" + params = { + "max_failures": 5, + "base_lock_seconds": 30, + "max_lock_seconds": 900, + "ttl_seconds": 3600, + "max_items": 4096, + "clock": clock, + } + params.update(overrides) + return LoginThrottle(**params) + + +KEY = ("10.0.0.1", "admin") + + +def test_key_normalizes_username_and_tolerates_missing_host() -> None: + """用户名去空白并小写，客户端地址缺失时用空串。""" + assert build_login_throttle_key("10.0.0.1", " Admin ") == ("10.0.0.1", "admin") + assert build_login_throttle_key(None, None) == ("", "") + + +def test_failures_below_threshold_do_not_lock() -> None: + """未达到阈值前不锁定。""" + throttle = _throttle(_Clock()) + for _ in range(4): + assert throttle.record_failure(KEY) == 0 + assert throttle.retry_after(KEY) == 0 + + +def test_reaching_threshold_locks_and_doubles_until_capped() -> None: + """第 5 次失败锁 30 秒，之后每次翻倍直到 900 秒上限。""" + clock = _Clock() + throttle = _throttle(clock) + for _ in range(4): + throttle.record_failure(KEY) + + assert throttle.record_failure(KEY) == 30 + assert throttle.retry_after(KEY) == 30 + + expected = [60, 120, 240, 480, 900, 900] + for lock_seconds in expected: + assert throttle.record_failure(KEY) == lock_seconds + assert throttle.retry_after(KEY) == lock_seconds + + +def test_retry_after_counts_down_and_releases() -> None: + """锁定期内返回剩余秒数（向上取整），到期后放行。""" + clock = _Clock() + throttle = _throttle(clock) + for _ in range(5): + throttle.record_failure(KEY) + + clock.advance(10.2) + assert throttle.retry_after(KEY) == 20 + clock.advance(20) + assert throttle.retry_after(KEY) == 0 + + +def test_success_clears_failure_history() -> None: + """登录成功后计数清零，再失败从头计数。""" + throttle = _throttle(_Clock()) + for _ in range(5): + throttle.record_failure(KEY) + + throttle.record_success(KEY) + + assert throttle.retry_after(KEY) == 0 + assert throttle.record_failure(KEY) == 0 + + +def test_stale_entries_expire_after_ttl() -> None: + """超过 TTL 且不在锁定期的记录被清理，再失败从头计数。""" + clock = _Clock() + throttle = _throttle(clock, ttl_seconds=60) + for _ in range(3): + throttle.record_failure(KEY) + + clock.advance(61) + throttle.record_failure(("10.0.0.2", "other")) + + assert throttle.record_failure(KEY) == 0 + + +def test_capacity_evicts_least_recently_updated() -> None: + """超过容量时淘汰最久未更新的键。""" + clock = _Clock() + throttle = _throttle(clock, max_items=2) + throttle.record_failure(("1.1.1.1", "a")) + clock.advance(1) + throttle.record_failure(("2.2.2.2", "b")) + clock.advance(1) + throttle.record_failure(("3.3.3.3", "c")) + + assert ("1.1.1.1", "a") not in throttle._entries # pylint: disable=protected-access + assert ("2.2.2.2", "b") in throttle._entries # pylint: disable=protected-access + assert ("3.3.3.3", "c") in throttle._entries # pylint: disable=protected-access + + +def test_keys_are_isolated_by_client_and_username() -> None: + """不同客户端或不同用户名互不影响。""" + throttle = _throttle(_Clock()) + for _ in range(5): + throttle.record_failure(KEY) + + assert throttle.retry_after(("10.0.0.2", "admin")) == 0 + assert throttle.retry_after(("10.0.0.1", "other")) == 0 + + +def test_default_throttle_is_shared() -> None: + """进程内默认实例唯一。""" + assert get_login_throttle() is get_login_throttle()