腾讯WorkBuddy
7f1a93f6a6
- 公司账号创建改为随机一次性初始密码，只在创建响应中显示一次， 密码保证不等于用户名；删除「用户名即初始密码」兼容分支，继续 强制首次登录改密。 - 跨公司相同字节文件上传只返回通用重复状态：不再返回其他公司的 原批次 ID、银行、模板、期间、交易数或诊断；同公司重复上传的 幂等摘要保持可用。 - 补充服务端回归测试，覆盖同公司与跨公司两个分支及随机密码； 完整测试 85 项全绿，node --check 通过。
253 lines
11 KiB
Python
253 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
import sqlite3
|
|
import unittest
|
|
|
|
from bank_importer import auth
|
|
from bank_importer.db import connect, migrate, utc_now
|
|
|
|
|
|
class AuthTestCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.connection = connect(":memory:")
|
|
self.addCleanup(self.connection.close)
|
|
migrate(self.connection)
|
|
now = utc_now()
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
|
|
(now, now),
|
|
)
|
|
self.company_id = int(self.connection.execute("SELECT id FROM companies").fetchone()["id"])
|
|
|
|
def create_company_user(self, username: str = "cashier-a", password: str = "Init1234") -> int:
|
|
return auth.create_user(
|
|
self.connection, username, password, "company", company_id=self.company_id
|
|
)
|
|
|
|
def create_admin(self, username: str = "group-admin", password: str = "Admin1234") -> int:
|
|
return auth.create_user(self.connection, username, password, "admin")
|
|
|
|
|
|
class PasswordHashTests(unittest.TestCase):
|
|
def test_hash_format_and_verify_roundtrip(self) -> None:
|
|
stored = auth.hash_password("Secret123")
|
|
scheme, iterations, salt_hex, hash_hex = stored.split("$")
|
|
self.assertEqual("pbkdf2_sha256", scheme)
|
|
self.assertEqual(str(auth.PBKDF2_ITERATIONS), iterations)
|
|
self.assertEqual(32, len(salt_hex))
|
|
self.assertEqual(64, len(hash_hex))
|
|
self.assertTrue(auth.verify_password("Secret123", stored))
|
|
|
|
def test_wrong_password_is_rejected(self) -> None:
|
|
stored = auth.hash_password("Secret123")
|
|
self.assertFalse(auth.verify_password("Secret124", stored))
|
|
|
|
def test_same_password_gets_different_salts(self) -> None:
|
|
self.assertNotEqual(auth.hash_password("Secret123"), auth.hash_password("Secret123"))
|
|
|
|
def test_malformed_stored_hash_is_rejected(self) -> None:
|
|
for stored in ("", "plain", "pbkdf2_sha256$x$y$z", "bcrypt$1$aa$bb"):
|
|
with self.subTest(stored=stored):
|
|
self.assertFalse(auth.verify_password("Secret123", stored))
|
|
|
|
|
|
class PasswordPolicyTests(unittest.TestCase):
|
|
def test_policy_rejects_short_username_like_and_single_charset(self) -> None:
|
|
self.assertIsNotNone(auth.validate_password_policy("Ab1", "cashier"))
|
|
self.assertIsNotNone(auth.validate_password_policy("Cashier99", "cashier99"))
|
|
self.assertIsNotNone(auth.validate_password_policy("abcdefgh", "cashier"))
|
|
self.assertIsNotNone(auth.validate_password_policy("12345678", "cashier"))
|
|
self.assertIsNone(auth.validate_password_policy("GoodPass1", "cashier"))
|
|
|
|
def test_initial_password_generator_guarantees_charset(self) -> None:
|
|
for _ in range(50):
|
|
password = auth.generate_initial_password()
|
|
self.assertEqual(12, len(password))
|
|
self.assertTrue(any(char.isupper() for char in password))
|
|
self.assertTrue(any(char.islower() for char in password))
|
|
self.assertTrue(any(char.isdigit() for char in password))
|
|
self.assertIsNone(auth.validate_password_policy(auth.generate_initial_password(), "x"))
|
|
|
|
def test_initial_password_exclude_never_equals_username(self) -> None:
|
|
for _ in range(50):
|
|
password = auth.generate_initial_password(exclude="Cashier99")
|
|
self.assertNotEqual(password.lower(), "cashier99")
|
|
|
|
|
|
class CreateUserTests(AuthTestCase):
|
|
def test_company_role_requires_company(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
auth.create_user(self.connection, "cashier-x", "Init1234", "company")
|
|
|
|
def test_company_must_exist(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
auth.create_user(self.connection, "cashier-x", "Init1234", "company", company_id=999)
|
|
|
|
def test_admin_cannot_bind_company(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
auth.create_user(
|
|
self.connection, "admin-x", "Init1234", "admin", company_id=self.company_id
|
|
)
|
|
|
|
def test_invalid_role_rejected(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
auth.create_user(self.connection, "nobody", "Init1234", "superuser")
|
|
|
|
def test_duplicate_username_rejected(self) -> None:
|
|
self.create_company_user()
|
|
with self.assertRaises(ValueError):
|
|
self.create_company_user()
|
|
|
|
def test_new_company_user_must_change_password_by_default(self) -> None:
|
|
user_id = self.create_company_user()
|
|
user = self.connection.execute(
|
|
"SELECT must_change_password, status FROM users WHERE id = ?", (user_id,)
|
|
).fetchone()
|
|
self.assertEqual(1, user["must_change_password"])
|
|
self.assertEqual("active", user["status"])
|
|
|
|
|
|
class AuthenticateTests(AuthTestCase):
|
|
def test_success_returns_user_and_records_attempt(self) -> None:
|
|
self.create_company_user()
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Init1234", "127.0.0.1")
|
|
self.assertIsNotNone(user)
|
|
self.assertIsNone(reason)
|
|
attempt = self.connection.execute("SELECT success FROM login_attempts").fetchone()
|
|
self.assertEqual(1, attempt["success"])
|
|
entry = self.connection.execute(
|
|
"SELECT action FROM audit_log WHERE action = 'login_success'"
|
|
).fetchone()
|
|
self.assertIsNotNone(entry)
|
|
|
|
def test_bad_credentials_do_not_leak_which_part_failed(self) -> None:
|
|
self.create_company_user()
|
|
for username, password in (("cashier-a", "Wrong999"), ("no-such-user", "Init1234")):
|
|
with self.subTest(username=username):
|
|
user, reason = auth.authenticate(self.connection, username, password, "127.0.0.1")
|
|
self.assertIsNone(user)
|
|
self.assertEqual("bad_credentials", reason)
|
|
|
|
def test_disabled_user_gets_disabled_reason(self) -> None:
|
|
user_id = self.create_company_user()
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"UPDATE users SET status = 'disabled', updated_at = ? WHERE id = ?",
|
|
(utc_now(), user_id),
|
|
)
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Init1234", "127.0.0.1")
|
|
self.assertIsNone(user)
|
|
self.assertEqual("disabled", reason)
|
|
|
|
def test_five_failures_within_window_trigger_rate_limit(self) -> None:
|
|
self.create_company_user()
|
|
for _ in range(auth.RATE_LIMIT_MAX_FAILURES):
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Wrong999", "10.0.0.1")
|
|
self.assertEqual("bad_credentials", reason)
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Init1234", "10.0.0.1")
|
|
self.assertIsNone(user)
|
|
self.assertEqual("rate_limited", reason)
|
|
# A different IP is not limited.
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Init1234", "10.0.0.2")
|
|
self.assertIsNotNone(user)
|
|
|
|
def test_failures_outside_window_do_not_count(self) -> None:
|
|
self.create_company_user()
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat()
|
|
with self.connection:
|
|
for _ in range(auth.RATE_LIMIT_MAX_FAILURES):
|
|
self.connection.execute(
|
|
"INSERT INTO login_attempts (username, ip, success, created_at) VALUES (?, ?, 0, ?)",
|
|
("cashier-a", "10.0.0.9", old),
|
|
)
|
|
user, reason = auth.authenticate(self.connection, "cashier-a", "Init1234", "10.0.0.9")
|
|
self.assertIsNotNone(user)
|
|
self.assertIsNone(reason)
|
|
|
|
|
|
class SessionTests(AuthTestCase):
|
|
def test_create_and_resolve_roundtrip(self) -> None:
|
|
user_id = self.create_company_user()
|
|
token = auth.create_session(self.connection, user_id)
|
|
user = auth.resolve_session(self.connection, token)
|
|
self.assertIsNotNone(user)
|
|
self.assertEqual(user_id, user["id"])
|
|
# Only the digest is stored, never the raw token.
|
|
row = self.connection.execute("SELECT token_hash FROM sessions").fetchone()
|
|
self.assertEqual(hashlib.sha256(token.encode()).hexdigest(), row["token_hash"])
|
|
self.assertNotIn(token, row["token_hash"])
|
|
|
|
def test_expired_session_is_rejected(self) -> None:
|
|
user_id = self.create_company_user()
|
|
token = auth.create_session(self.connection, user_id)
|
|
past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"UPDATE sessions SET expires_at = ? WHERE token_hash = ?",
|
|
(past, hashlib.sha256(token.encode()).hexdigest()),
|
|
)
|
|
self.assertIsNone(auth.resolve_session(self.connection, token))
|
|
|
|
def test_revoked_session_is_rejected(self) -> None:
|
|
user_id = self.create_company_user()
|
|
token = auth.create_session(self.connection, user_id)
|
|
auth.revoke_session(self.connection, token)
|
|
self.assertIsNone(auth.resolve_session(self.connection, token))
|
|
|
|
def test_revoke_user_sessions_kills_all_sessions(self) -> None:
|
|
user_id = self.create_company_user()
|
|
first = auth.create_session(self.connection, user_id)
|
|
second = auth.create_session(self.connection, user_id)
|
|
auth.revoke_user_sessions(self.connection, user_id)
|
|
self.assertIsNone(auth.resolve_session(self.connection, first))
|
|
self.assertIsNone(auth.resolve_session(self.connection, second))
|
|
|
|
def test_disabled_user_session_is_rejected(self) -> None:
|
|
user_id = self.create_company_user()
|
|
token = auth.create_session(self.connection, user_id)
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"UPDATE users SET status = 'disabled', updated_at = ? WHERE id = ?",
|
|
(utc_now(), user_id),
|
|
)
|
|
self.assertIsNone(auth.resolve_session(self.connection, token))
|
|
|
|
|
|
class ChangePasswordTests(AuthTestCase):
|
|
def test_change_password_clears_flag_and_updates_hash(self) -> None:
|
|
user_id = self.create_company_user()
|
|
error = auth.change_password(self.connection, user_id, "Init1234", "NewPass99")
|
|
self.assertIsNone(error)
|
|
user = self.connection.execute(
|
|
"SELECT password_hash, must_change_password FROM users WHERE id = ?", (user_id,)
|
|
).fetchone()
|
|
self.assertEqual(0, user["must_change_password"])
|
|
self.assertTrue(auth.verify_password("NewPass99", user["password_hash"]))
|
|
self.assertFalse(auth.verify_password("Init1234", user["password_hash"]))
|
|
entry = self.connection.execute(
|
|
"SELECT detail FROM audit_log WHERE action = 'password_change'"
|
|
).fetchone()
|
|
self.assertIsNotNone(entry)
|
|
self.assertNotIn("NewPass99", entry["detail"] or "")
|
|
|
|
def test_wrong_old_password_rejected(self) -> None:
|
|
user_id = self.create_company_user()
|
|
error = auth.change_password(self.connection, user_id, "Wrong999", "NewPass99")
|
|
self.assertIsNotNone(error)
|
|
user = self.connection.execute(
|
|
"SELECT must_change_password FROM users WHERE id = ?", (user_id,)
|
|
).fetchone()
|
|
self.assertEqual(1, user["must_change_password"])
|
|
|
|
def test_policy_violation_rejected(self) -> None:
|
|
user_id = self.create_company_user()
|
|
error = auth.change_password(self.connection, user_id, "Init1234", "short")
|
|
self.assertIsNotNone(error)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|