HEL-166: 增加 APP_LOGIN_RATE_LIMIT_DISABLED 运维开关（默认保持限流）

Co-authored-by: multica-agent <github@multica.ai>

README.md

@@ -45,6 +45,8 @@ SHA-256 内容哈希不可变保存，重复上传返回 `duplicate` 状态并
   公司账号创建时生成随机一次性初始密码，只在创建响应中显示一次，
   首次登录强制改密（「重置密码」同样生成随机一次性密码并吊销会话）。
 - 会话有效期 8 小时；同一账号同一 IP 10 分钟内登录失败 5 次将被限流。
+  内网测试环境可设 `APP_LOGIN_RATE_LIMIT_DISABLED=1` 暂时关闭该锁定
+  （默认未设置 = 保持限流，正式环境不得开启该变量）。
 
 访问地址（服务默认监听 `0.0.0.0:4173`，同局域网设备把 `127.0.0.1` 换成本机局域网 IP 即可访问；可用环境变量 `APP_HOST` / `APP_PORT` 覆盖）：
 

docs/decisions/003-auth.md

@@ -26,6 +26,9 @@
 - 计数来自 `login_attempts` 表，窗口为滚动 10 分钟；触发后返回 429，
   且在窗口内不再记录新尝试，行为确定、可测试。
 - 失败提示统一为「账号或密码不正确」，不泄露是哪一部分错误。
+- 运维开关：环境变量 `APP_LOGIN_RATE_LIMIT_DISABLED` 设为 `1/true/yes/on`
+  时完全跳过限流检查，仅限内网测试环境临时使用；默认未设置，保持
+  「5 次失败锁 10 分钟」的生产安全策略不变。
 
 ## 角色与公司绑定
 

src/bank_importer/auth.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 import hashlib
 import hmac
+import os
 import secrets
 import sqlite3
 import string
@@ -28,6 +29,17 @@ RATE_LIMIT_WINDOW_MINUTES = 10
 INITIAL_PASSWORD_LENGTH = 12
 
 
+def _env_flag(value: str | None) -> bool:
+    """Parse a yes/no style environment flag; blank/absent means False."""
+    return (value or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+# Ops switch for internal test environments: APP_LOGIN_RATE_LIMIT_DISABLED=1
+# turns off the login-failure lockout entirely. The default (unset) keeps the
+# production policy — 5 failures within 10 minutes lock the (账号, IP) pair.
+RATE_LIMIT_DISABLED = _env_flag(os.environ.get("APP_LOGIN_RATE_LIMIT_DISABLED"))
+
+
 def hash_password(password: str) -> str:
     """Hash ``password`` as ``pbkdf2_sha256$<iterations>$<salt_hex>$<hash_hex>``."""
     salt = secrets.token_bytes(16)
@@ -146,18 +158,19 @@ def authenticate(
     Every non-rate-limited attempt is recorded in ``login_attempts`` and
     ``audit_log``; the password itself is never stored anywhere.
     """
-    window_start = (
-        datetime.now(timezone.utc) - timedelta(minutes=RATE_LIMIT_WINDOW_MINUTES)
-    ).isoformat()
-    failures = connection.execute(
-        """
-        SELECT COUNT(*) AS n FROM login_attempts
-        WHERE username = ? AND ip = ? AND success = 0 AND created_at >= ?
-        """,
-        (username, ip, window_start),
-    ).fetchone()
-    if failures["n"] >= RATE_LIMIT_MAX_FAILURES:
-        return None, "rate_limited"
+    if not RATE_LIMIT_DISABLED:
+        window_start = (
+            datetime.now(timezone.utc) - timedelta(minutes=RATE_LIMIT_WINDOW_MINUTES)
+        ).isoformat()
+        failures = connection.execute(
+            """
+            SELECT COUNT(*) AS n FROM login_attempts
+            WHERE username = ? AND ip = ? AND success = 0 AND created_at >= ?
+            """,
+            (username, ip, window_start),
+        ).fetchone()
+        if failures["n"] >= RATE_LIMIT_MAX_FAILURES:
+            return None, "rate_limited"
 
     user = connection.execute(
         "SELECT * FROM users WHERE username = ?", (username,)

tests/test_auth.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta, timezone
 import hashlib
 import sqlite3
 import unittest
+from unittest import mock
 
 from bank_importer import auth
 from bank_importer.db import connect, migrate, utc_now
@@ -167,6 +168,40 @@ class AuthenticateTests(AuthTestCase):
         self.assertIsNotNone(user)
         self.assertIsNone(reason)
 
+    def test_env_flag_parses_truthy_and_falsy_values(self) -> None:
+        cases = {
+            "1": True,
+            "true": True,
+            "YES": True,
+            " on ": True,
+            "": False,
+            "0": False,
+            "false": False,
+            "off": False,
+            None: False,
+        }
+        for value, expected in cases.items():
+            with self.subTest(value=value):
+                self.assertEqual(expected, auth._env_flag(value))
+
+    def test_rate_limit_enabled_by_default(self) -> None:
+        self.assertFalse(auth.RATE_LIMIT_DISABLED)
+
+    def test_env_switch_disables_rate_limit_but_not_credential_checks(self) -> None:
+        self.create_company_user()
+        with mock.patch.object(auth, "RATE_LIMIT_DISABLED", True):
+            for _ in range(auth.RATE_LIMIT_MAX_FAILURES + 2):
+                user, reason = auth.authenticate(
+                    self.connection, "cashier-a", "Wrong999", "10.0.0.1"
+                )
+                self.assertIsNone(user)
+                self.assertEqual("bad_credentials", reason)
+            user, reason = auth.authenticate(
+                self.connection, "cashier-a", "Init1234", "10.0.0.1"
+            )
+            self.assertIsNotNone(user)
+            self.assertIsNone(reason)
+
 
 class SessionTests(AuthTestCase):
     def test_create_and_resolve_roundtrip(self) -> None: