Merge pull request #6925 from cwd0204/fix/login-throttle

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(

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

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",

app/locales/zh-TW.json

@@ -457,6 +457,7 @@
     "用户权限不足": "使用者權限不足",
     "用户名或密码错误": "使用者名稱或密碼錯誤",
     "需要二次验证": "需要二次驗證",
+    "登录尝试过于频繁，请稍后重试": "登入嘗試過於頻繁，請稍後重試",
     "图片读取出错": "圖片讀取出錯",
     "授权失败": "授權失敗",
     "报文内容为空": "報文內容為空",

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` 再执行快照测试及一次

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 等基线均为零 |

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.

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",

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 == []

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()