HEL-171 部署合并：公司端工作台待办同步(bf5754e) + 登录限流运维开关(551909a)

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