fix: 登录白名单 IP 绕过防暴破锁定

白名单内 IP 跳过失败次数锁定；失败仅审计不计数；成功登录清除该用户历史失败记录。

Co-authored-by: Cursor <cursoragent@cursor.com>

docs/changelog/2026-06-02-whitelist-bypass-lockout.md

@@ -0,0 +1,30 @@
+# Changelog: 登录白名单 IP 绕过防暴破锁定
+
+**日期**: 2026-06-02
+
+## 摘要
+
+修复登录流程中「白名单 IP 仍受 15 分钟失败锁定」的问题：白名单内 IP 跳过 `count_recent_failures` 锁定检查；失败时不写入 `login_attempts` 失败计数，仅写审计日志；成功登录时清除该用户历史失败记录，避免被其他 IP 的累计失败连带锁住。
+
+## 动机
+
+运维/内网可信 IP 已在登录白名单中，不应再被全局用户名维度的失败次数锁定；同时白名单 IP 登录成功后应解除因外网误试导致的账户锁定状态。
+
+## 涉及文件
+
+- `server/application/services/auth_service.py` — `login()` 增加 `is_whitelisted_ip`、`_handle_login_failure()`、成功时 `clear_failures_for_username`
+- `server/infrastructure/database/admin_repo.py` — `LoginAttemptRepositoryImpl.clear_failures_for_username`
+- `server/domain/repositories/__init__.py` — `LoginAttemptRepository` 协议补充方法
+
+## 迁移 / 重启
+
+- 无需数据库迁移
+- 需重启 Nexus 后端（`supervisorctl restart nexus`）后生效
+
+## 验证方式
+
+1. 本地：`python -c "import server.application.services.auth_service"`
+2. 开启 `LOGIN_ALLOWLIST_ENABLED`，将当前 IP 加入白名单
+3. 用非白名单 IP 对同一用户连续失败 5 次至锁定
+4. 从白名单 IP 登录：应不被 `account_locked` 拦截；密码错误时 `audit_logs` 有 `login_failed`，`login_attempts` 不新增失败行
+5. 白名单 IP 正确密码登录后，该用户 `login_attempts` 中 `success=0` 记录应被清除

server/application/services/auth_service.py

@@ -98,6 +98,7 @@ class AuthService:
         # ── IP allowlist check ──
         # Only enforced when login_allowlist_enabled = "true"
         allowlist_on = (getattr(settings, "LOGIN_ALLOWLIST_ENABLED", "false") or "false").lower()
+        is_whitelisted_ip = False
         if allowlist_on in ("true", "1", "yes", "on") and ip_address and ip_address not in ("", "unknown"):
             # Combine subscription IPs + manual IPs
             sub_ips_raw = (getattr(settings, "LOGIN_SUBSCRIPTION_IPS", "") or "").strip()
@@ -115,29 +116,38 @@ class AuthService:
                     "reason": "ip_blocked",
                     "message": f"当前 IP ({ip_address}) 不在登录白名单中，请检查代理或联系管理员",
                 }
+            if allowed_ips and _ip_in_allowlist(ip_address, allowed_ips):
+                is_whitelisted_ip = True
 
-        # Check brute-force lockout
-        failures = await self.attempt_repo.count_recent_failures(username, ip_address, LOCKOUT_MINUTES)
-        if failures >= MAX_LOGIN_FAILURES:
-            await self._audit(
-                "login_locked", "admin", 0, f"账号已锁定: {username}",
-                ip_address=ip_address, admin_username=username,
-            )
-            return {"success": False, "reason": "account_locked", "message": "登录尝试过多，请15分钟后重试"}
+        # Check brute-force lockout (trusted whitelist IPs skip lockout counting)
+        if not is_whitelisted_ip:
+            failures = await self.attempt_repo.count_recent_failures(username, ip_address, LOCKOUT_MINUTES)
+            if failures >= MAX_LOGIN_FAILURES:
+                await self._audit(
+                    "login_locked", "admin", 0, f"账号已锁定: {username}",
+                    ip_address=ip_address, admin_username=username,
+                )
+                return {"success": False, "reason": "account_locked", "message": "登录尝试过多，请15分钟后重试"}
 
         # Find admin
         admin = await self.admin_repo.get_by_username(username)
         if not admin:
-            await self._record_attempt(username, ip_address, False)
+            await self._handle_login_failure(
+                username, ip_address, is_whitelisted_ip, "login_failed", "用户名不存在",
+            )
             return {"success": False, "reason": "invalid_credentials", "message": "用户名或密码错误"}
 
         if not admin.is_active:
-            await self._record_attempt(username, ip_address, False)
+            await self._handle_login_failure(
+                username, ip_address, is_whitelisted_ip, "login_failed", "账户已禁用",
+            )
             return {"success": False, "reason": "account_disabled", "message": "账户已被禁用"}
 
         # Verify password
         if not self._verify_password(password, admin.password_hash):
-            await self._record_attempt(username, ip_address, False)
+            await self._handle_login_failure(
+                username, ip_address, is_whitelisted_ip, "login_failed", "密码错误",
+            )
             return {"success": False, "reason": "invalid_credentials", "message": "用户名或密码错误"}
 
         # Verify TOTP (if enabled)
@@ -145,7 +155,9 @@ class AuthService:
             if not totp_code:
                 return {"success": False, "reason": "totp_required", "message": "需要TOTP验证码"}
             if not self._verify_totp(totp_code, admin.totp_secret):
-                await self._record_attempt(username, ip_address, False)
+                await self._handle_login_failure(
+                    username, ip_address, is_whitelisted_ip, "login_failed", "TOTP验证码错误",
+                )
                 return {"success": False, "reason": "invalid_totp", "message": "TOTP验证码错误"}
 
         # Success — update admin first (so JWT captures latest updated_at/token_version)
@@ -159,6 +171,8 @@ class AuthService:
         # Store refresh token in Redis (supports multi-device login)
         await self._store_refresh_token(admin.id, refresh_token)
 
+        if is_whitelisted_ip:
+            await self.attempt_repo.clear_failures_for_username(username)
         await self._record_attempt(username, ip_address, True)
         await self._audit(
             "login_success", "admin", admin.id, f"登录: {username}",
@@ -507,6 +521,24 @@ class AuthService:
                 return True
         return False
 
+    async def _handle_login_failure(
+        self,
+        username: str,
+        ip_address: str,
+        is_whitelisted_ip: bool,
+        audit_action: str,
+        audit_detail: str,
+    ) -> None:
+        """Record failed login: audit always for whitelist IPs; brute-force count only otherwise."""
+        if is_whitelisted_ip:
+            await self._audit(
+                audit_action, "admin", 0,
+                f"{audit_detail} (用户: {username}, 白名单IP)",
+                ip_address=ip_address, admin_username=username,
+            )
+        else:
+            await self._record_attempt(username, ip_address, False)
+
     async def _record_attempt(self, username: str, ip_address: str, success: bool):
         """Record login attempt for brute-force tracking"""
         attempt = LoginAttempt(username=username, ip_address=ip_address, success=success)

server/domain/repositories/__init__.py

@@ -47,6 +47,7 @@ class LoginAttemptRepository(Protocol):
 
     async def create(self, attempt: LoginAttempt) -> LoginAttempt: ...
     async def count_recent_failures(self, username: str, ip_address: str, minutes: int = 15) -> int: ...
+    async def clear_failures_for_username(self, username: str) -> int: ...
 
 
 class SettingRepository(Protocol):

server/infrastructure/database/admin_repo.py

@@ -1,7 +1,7 @@
 """Nexus — Admin & LoginAttempt Repository (Async SQLAlchemy)"""
 
 from typing import Optional
-from sqlalchemy import select, func
+from sqlalchemy import select, func, delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from server.domain.models import Admin, LoginAttempt
@@ -50,4 +50,13 @@ class LoginAttemptRepositoryImpl:
             .where(LoginAttempt.success == False)
             .where(LoginAttempt.attempted_at >= cutoff)
         )
-        return result.scalar_one() or 0
+        return result.scalar_one() or 0
+
+    async def clear_failures_for_username(self, username: str) -> int:
+        result = await self.session.execute(
+            delete(LoginAttempt)
+            .where(LoginAttempt.username == username)
+            .where(LoginAttempt.success == False)
+        )
+        await self.session.commit()
+        return result.rowcount or 0