From 551909ad0c34a4a3d1e167e2c9367670f57ea28c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Thu, 27 Aug 2026 11:48:00 +0000 Subject: [PATCH] =?UTF-8?q?HEL-166:=20=E5=A2=9E=E5=8A=A0=20APP=5FLOGIN=5FR?= =?UTF-8?q?ATE=5FLIMIT=5FDISABLED=20=E8=BF=90=E7=BB=B4=E5=BC=80=E5=85=B3?= =?UTF-8?q?=EF=BC=88=E9=BB=98=E8=AE=A4=E4=BF=9D=E6=8C=81=E9=99=90=E6=B5=81?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: multica-agent --- README.md | 2 ++ docs/decisions/003-auth.md | 3 +++ src/bank_importer/auth.py | 37 +++++++++++++++++++++++++------------ tests/test_auth.py | 35 +++++++++++++++++++++++++++++++++++ 4 files changed, 65 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index ac40353..1156327 100644 --- a/README.md +++ b/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` 覆盖）： diff --git a/docs/decisions/003-auth.md b/docs/decisions/003-auth.md index f62df1b..9e9fad7 100644 --- a/docs/decisions/003-auth.md +++ b/docs/decisions/003-auth.md @@ -26,6 +26,9 @@ - 计数来自 `login_attempts` 表，窗口为滚动 10 分钟；触发后返回 429， 且在窗口内不再记录新尝试，行为确定、可测试。 - 失败提示统一为「账号或密码不正确」，不泄露是哪一部分错误。 +- 运维开关：环境变量 `APP_LOGIN_RATE_LIMIT_DISABLED` 设为 `1/true/yes/on` + 时完全跳过限流检查，仅限内网测试环境临时使用；默认未设置，保持 + 「5 次失败锁 10 分钟」的生产安全策略不变。 ## 角色与公司绑定 diff --git a/src/bank_importer/auth.py b/src/bank_importer/auth.py index 37445ec..36fffd4 100644 --- a/src/bank_importer/auth.py +++ b/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$$$``.""" 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,) diff --git a/tests/test_auth.py b/tests/test_auth.py index 74188b7..f8cd503 100644 --- a/tests/test_auth.py +++ b/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: