B-40: 正式登录与租户隔离——认证、会话、权限与审计
- 管理员/公司两类角色：PBKDF2 密码散列、随机初始密码、首次改密、 停用、重置密码与吊销会话。 - 会话只存令牌摘要，8 小时绝对过期；登录失败同 (账号,IP) 限流。 - 公司账号服务端绑定唯一公司；读取/上传/导出/主数据/审核接口逐项 服务端授权，跨公司访问返回 404 而非 403。 - 审计日志记录登录、改密、上传、导出、建公司、建/停/启用户。 - 决策记录见 docs/decisions/003-auth.md。
This commit is contained in:
腾讯WorkBuddy
@@ -0,0 +1,247 @@
|
||||
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"))
|
||||
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,521 @@
|
||||
"""HTTP integration tests for authentication, RBAC and tenant isolation.
|
||||
|
||||
Spins up a real ``ThreadingHTTPServer`` with a temp database/storage and
|
||||
drives it with stdlib ``http.client`` (cookies handled by hand). The server
|
||||
module reads ``APP_DB_PATH`` / ``APP_STORAGE_DIR`` from module globals at
|
||||
request time, so tests patch them per class.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from http.client import HTTPConnection
|
||||
from http.cookies import SimpleCookie
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
from bank_importer import auth
|
||||
from bank_importer.db import connect, migrate, utc_now
|
||||
|
||||
import server
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SAMPLES = ROOT / "流水模板"
|
||||
CCB_SAMPLE = SAMPLES / "中国建设银行账户流水.xls"
|
||||
CITIC_SAMPLE = SAMPLES / "中信银行账户流水.xlsx"
|
||||
|
||||
BOOTSTRAP_PASSWORD = "BootAdmin123"
|
||||
ADMIN_PASSWORD = "AdminPass123"
|
||||
CASHIER_A_PASSWORD = "CashierA123"
|
||||
|
||||
|
||||
class Client:
|
||||
"""Minimal HTTP client with a cookie jar."""
|
||||
|
||||
def __init__(self, host: str, port: int) -> None:
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.cookies: dict[str, str] = {}
|
||||
|
||||
def request(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
body: bytes | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
) -> tuple[int, dict[str, str], bytes]:
|
||||
connection = HTTPConnection(self.host, self.port)
|
||||
request_headers = dict(headers or {})
|
||||
if self.cookies:
|
||||
request_headers["Cookie"] = "; ".join(
|
||||
f"{key}={value}" for key, value in self.cookies.items()
|
||||
)
|
||||
connection.request(method, path, body=body, headers=request_headers)
|
||||
response = connection.getresponse()
|
||||
data = response.read()
|
||||
response_headers = {key.lower(): value for key, value in response.getheaders()}
|
||||
set_cookie = response_headers.get("set-cookie")
|
||||
if set_cookie:
|
||||
cookie = SimpleCookie()
|
||||
cookie.load(set_cookie)
|
||||
for key, morsel in cookie.items():
|
||||
if morsel.value:
|
||||
self.cookies[key] = morsel.value
|
||||
else:
|
||||
self.cookies.pop(key, None)
|
||||
status = response.status
|
||||
connection.close()
|
||||
return status, response_headers, data
|
||||
|
||||
def get(self, path: str) -> tuple[int, dict[str, str], bytes]:
|
||||
return self.request("GET", path)
|
||||
|
||||
def post_json(self, path: str, payload: dict) -> tuple[int, dict, bytes]:
|
||||
return self.request(
|
||||
"POST",
|
||||
path,
|
||||
body=json.dumps(payload).encode("utf-8"),
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
|
||||
def post_multipart(
|
||||
self, path: str, fields: dict[str, str], filename: str, content: bytes
|
||||
) -> tuple[int, dict, bytes]:
|
||||
boundary = "----cwtestboundary7f3a9c1e"
|
||||
parts: list[bytes] = []
|
||||
for name, value in fields.items():
|
||||
parts.append(
|
||||
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode()
|
||||
)
|
||||
parts.append(
|
||||
f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n'
|
||||
"Content-Type: application/octet-stream\r\n\r\n".encode()
|
||||
+ content
|
||||
+ b"\r\n"
|
||||
)
|
||||
parts.append(f"--{boundary}--\r\n".encode())
|
||||
return self.request(
|
||||
"POST",
|
||||
path,
|
||||
body=b"".join(parts),
|
||||
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
||||
)
|
||||
|
||||
|
||||
def as_json(data: bytes) -> dict:
|
||||
return json.loads(data.decode("utf-8"))
|
||||
|
||||
|
||||
class ServerAuthMatrixTests(unittest.TestCase):
|
||||
"""One live server; setUpClass builds the shared fixture via the API."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.temp_dir = tempfile.TemporaryDirectory()
|
||||
root = Path(cls.temp_dir.name)
|
||||
cls.db_path = root / "app.db"
|
||||
cls.storage = root / "files"
|
||||
|
||||
cls._old_db_path = server.DB_PATH
|
||||
cls._old_storage = server.STORAGE_DIR
|
||||
server.DB_PATH = cls.db_path
|
||||
server.STORAGE_DIR = cls.storage
|
||||
|
||||
os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
|
||||
connection = connect(cls.db_path)
|
||||
migrate(connection)
|
||||
generated = server.ensure_bootstrap_admin(connection)
|
||||
assert generated is None, "env password set, nothing should be generated"
|
||||
connection.close()
|
||||
|
||||
class QuietHandler(server.AppHandler):
|
||||
def log_message(self, *args) -> None: # silence per-request logs
|
||||
pass
|
||||
|
||||
cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
|
||||
cls.port = cls.httpd.server_address[1]
|
||||
cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
|
||||
cls.thread.start()
|
||||
|
||||
cls.known_passwords = {BOOTSTRAP_PASSWORD, ADMIN_PASSWORD, CASHIER_A_PASSWORD}
|
||||
|
||||
# --- Admin bootstrap: must_change_password gate, then change. ---
|
||||
cls.admin = Client("127.0.0.1", cls.port)
|
||||
status, _, data = cls.admin.post_json(
|
||||
"/api/login",
|
||||
{"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
|
||||
)
|
||||
assert status == 200, data
|
||||
assert as_json(data)["must_change_password"] is True
|
||||
status, _, data = cls.admin.get("/api/batches")
|
||||
assert status == 403, "must_change_password must block API access"
|
||||
status, _, data = cls.admin.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
|
||||
)
|
||||
assert status == 200, data
|
||||
status, _, _ = cls.admin.get("/api/batches")
|
||||
assert status == 200
|
||||
|
||||
# --- Companies A and B. ---
|
||||
status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "甲公司"})
|
||||
assert status == 200, data
|
||||
cls.company_a = as_json(data)["company_id"]
|
||||
status, _, data = cls.admin.post_json("/api/admin/companies", {"name": "乙公司"})
|
||||
assert status == 200, data
|
||||
cls.company_b = as_json(data)["company_id"]
|
||||
|
||||
# --- Company user A: initial password shown once, forced change. ---
|
||||
status, _, data = cls.admin.post_json(
|
||||
"/api/admin/users", {"username": "cashier-a", "company_id": cls.company_a}
|
||||
)
|
||||
assert status == 200, data
|
||||
payload = as_json(data)
|
||||
# Product decision: registration initial password equals the username.
|
||||
assert payload["initial_password"] == "cashier-a"
|
||||
cls.initial_password_a = payload["initial_password"]
|
||||
cls.known_passwords.add(cls.initial_password_a)
|
||||
|
||||
cls.cashier_a = Client("127.0.0.1", cls.port)
|
||||
status, _, data = cls.cashier_a.post_json(
|
||||
"/api/login",
|
||||
{
|
||||
"username": "cashier-a",
|
||||
"password": cls.initial_password_a,
|
||||
"portal": "company",
|
||||
},
|
||||
)
|
||||
assert status == 200, data
|
||||
assert as_json(data)["must_change_password"] is True
|
||||
status, _, _ = cls.cashier_a.get("/api/batches")
|
||||
assert status == 403, "must_change_password must block company API access"
|
||||
status, _, data = cls.cashier_a.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": cls.initial_password_a, "new_password": CASHIER_A_PASSWORD},
|
||||
)
|
||||
assert status == 200, data
|
||||
|
||||
# --- B gets a batch (admin upload), A gets its own batch. ---
|
||||
status, _, data = cls.admin.post_multipart(
|
||||
"/api/parse",
|
||||
{"company_id": str(cls.company_b)},
|
||||
CCB_SAMPLE.name,
|
||||
CCB_SAMPLE.read_bytes(),
|
||||
)
|
||||
assert status == 200, data
|
||||
cls.b_batch_id = as_json(data)["batch_id"]
|
||||
|
||||
status, _, data = cls.cashier_a.post_multipart(
|
||||
"/api/parse", {}, CITIC_SAMPLE.name, CITIC_SAMPLE.read_bytes()
|
||||
)
|
||||
assert status == 200, data
|
||||
cls.a_batch_id = as_json(data)["batch_id"]
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
cls.httpd.shutdown()
|
||||
cls.httpd.server_close()
|
||||
server.DB_PATH = cls._old_db_path
|
||||
server.STORAGE_DIR = cls._old_storage
|
||||
os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
|
||||
cls.temp_dir.cleanup()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def fresh_client(self) -> Client:
|
||||
return Client("127.0.0.1", self.port)
|
||||
|
||||
def create_company_user(self, username: str) -> tuple[Client, str, int]:
|
||||
status, _, data = self.admin.post_json(
|
||||
"/api/admin/users", {"username": username, "company_id": self.company_a}
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
payload = as_json(data)
|
||||
initial = payload["initial_password"]
|
||||
self.known_passwords.add(initial)
|
||||
client = self.fresh_client()
|
||||
status, _, data = client.post_json(
|
||||
"/api/login",
|
||||
{"username": username, "password": initial, "portal": "company"},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
new_password = "Changed123"
|
||||
self.known_passwords.add(new_password)
|
||||
status, _, data = client.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": initial, "new_password": new_password},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
return client, new_password, payload["user_id"]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Unauthenticated matrix
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_unauthenticated_api_calls_return_401(self) -> None:
|
||||
anon = self.fresh_client()
|
||||
for method_check in (
|
||||
lambda: anon.post_multipart("/api/parse", {}, "x.xls", b"data"),
|
||||
lambda: anon.get("/api/batches"),
|
||||
lambda: anon.get(f"/api/batches/{self.a_batch_id}/rows"),
|
||||
lambda: anon.get("/api/export.csv"),
|
||||
lambda: anon.get("/api/admin/users"),
|
||||
lambda: anon.get("/api/admin/companies"),
|
||||
lambda: anon.get("/api/admin/audit-log"),
|
||||
lambda: anon.get("/api/me"),
|
||||
):
|
||||
status, _, data = method_check()
|
||||
self.assertEqual(401, status, data)
|
||||
self.assertEqual("error", as_json(data)["status"])
|
||||
|
||||
def test_unauthenticated_portal_pages_redirect(self) -> None:
|
||||
anon = self.fresh_client()
|
||||
for page in ("/admin.html", "/company.html"):
|
||||
status, headers, _ = anon.get(page)
|
||||
self.assertEqual(302, status, page)
|
||||
self.assertEqual("/", headers.get("location"))
|
||||
|
||||
def test_wrong_password_returns_generic_401(self) -> None:
|
||||
anon = self.fresh_client()
|
||||
status, _, data = anon.post_json(
|
||||
"/api/login",
|
||||
{"username": "group-admin", "password": "WrongPass1", "portal": "admin"},
|
||||
)
|
||||
self.assertEqual(401, status)
|
||||
message = as_json(data)["message"]
|
||||
self.assertNotIn("密码不正确", message.replace("账号或密码不正确", ""))
|
||||
|
||||
def test_rate_limit_after_five_failures(self) -> None:
|
||||
anon = self.fresh_client()
|
||||
for _ in range(5):
|
||||
status, _, _ = anon.post_json(
|
||||
"/api/login",
|
||||
{"username": "ghost-user", "password": "WrongPass1", "portal": "admin"},
|
||||
)
|
||||
self.assertEqual(401, status)
|
||||
status, _, _ = anon.post_json(
|
||||
"/api/login",
|
||||
{"username": "ghost-user", "password": "WrongPass1", "portal": "admin"},
|
||||
)
|
||||
self.assertEqual(429, status)
|
||||
|
||||
def test_portal_mismatch_returns_403(self) -> None:
|
||||
anon = self.fresh_client()
|
||||
status, _, data = anon.post_json(
|
||||
"/api/login",
|
||||
{"username": "group-admin", "password": ADMIN_PASSWORD, "portal": "company"},
|
||||
)
|
||||
self.assertEqual(403, status)
|
||||
self.assertIn("端口", as_json(data)["message"])
|
||||
|
||||
def test_logout_revokes_session(self) -> None:
|
||||
client, _, _ = self.create_company_user("cashier-logout")
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(200, status)
|
||||
status, _, _ = client.request("POST", "/api/logout")
|
||||
self.assertEqual(200, status)
|
||||
self.assertNotIn("cw_session", client.cookies)
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(401, status)
|
||||
|
||||
def test_expired_session_returns_401(self) -> None:
|
||||
client, _, user_id = self.create_company_user("cashier-expired")
|
||||
connection = connect(self.db_path)
|
||||
try:
|
||||
with connection:
|
||||
connection.execute(
|
||||
"UPDATE sessions SET expires_at = ? WHERE user_id = ?",
|
||||
("2000-01-01T00:00:00+00:00", user_id),
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(401, status)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Admin and company lifecycle
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_duplicate_company_name_returns_409(self) -> None:
|
||||
status, _, _ = self.admin.post_json("/api/admin/companies", {"name": "甲公司"})
|
||||
self.assertEqual(409, status)
|
||||
|
||||
def test_me_returns_profile_without_password_material(self) -> None:
|
||||
status, _, data = self.cashier_a.get("/api/me")
|
||||
self.assertEqual(200, status)
|
||||
payload = as_json(data)
|
||||
self.assertEqual("cashier-a", payload["username"])
|
||||
self.assertEqual("company", payload["role"])
|
||||
self.assertEqual(self.company_a, payload["company_id"])
|
||||
self.assertEqual("甲公司", payload["company_name"])
|
||||
self.assertFalse(payload["must_change_password"])
|
||||
|
||||
def test_company_user_forbidden_on_all_admin_endpoints(self) -> None:
|
||||
calls = (
|
||||
lambda: self.cashier_a.get("/api/admin/companies"),
|
||||
lambda: self.cashier_a.post_json("/api/admin/companies", {"name": "丙公司"}),
|
||||
lambda: self.cashier_a.get("/api/admin/users"),
|
||||
lambda: self.cashier_a.post_json(
|
||||
"/api/admin/users", {"username": "x", "company_id": self.company_a}
|
||||
),
|
||||
lambda: self.cashier_a.request("POST", "/api/admin/users/1/disable"),
|
||||
lambda: self.cashier_a.request("POST", "/api/admin/users/1/enable"),
|
||||
lambda: self.cashier_a.request("POST", "/api/admin/users/1/reset-password"),
|
||||
lambda: self.cashier_a.get("/api/admin/audit-log"),
|
||||
)
|
||||
for call in calls:
|
||||
status, _, data = call()
|
||||
self.assertEqual(403, status, data)
|
||||
|
||||
def test_admin_upload_requires_company_id(self) -> None:
|
||||
status, _, data = self.admin.post_multipart(
|
||||
"/api/parse", {}, CCB_SAMPLE.name, CCB_SAMPLE.read_bytes()
|
||||
)
|
||||
self.assertEqual(400, status, data)
|
||||
|
||||
def test_admin_batches_filter_by_company(self) -> None:
|
||||
status, _, data = self.admin.get(f"/api/batches?company_id={self.company_b}")
|
||||
self.assertEqual(200, status)
|
||||
batches = as_json(data)["batches"]
|
||||
self.assertTrue(batches)
|
||||
for batch in batches:
|
||||
self.assertEqual(self.company_b, batch["company_id"])
|
||||
self.assertEqual("乙公司", batch["company_name"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Tenant isolation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_batches_scoped_to_own_company(self) -> None:
|
||||
status, _, data = self.cashier_a.get("/api/batches")
|
||||
self.assertEqual(200, status)
|
||||
batches = as_json(data)["batches"]
|
||||
self.assertTrue(batches)
|
||||
for batch in batches:
|
||||
self.assertEqual(self.company_a, batch["company_id"])
|
||||
self.assertNotIn(self.b_batch_id, [batch["id"] for batch in batches])
|
||||
|
||||
def test_idor_batch_rows_of_other_company_return_404(self) -> None:
|
||||
status, _, _ = self.cashier_a.get(f"/api/batches/{self.b_batch_id}/rows")
|
||||
self.assertEqual(404, status)
|
||||
status, _, data = self.cashier_a.get(f"/api/batches/{self.a_batch_id}/rows")
|
||||
self.assertEqual(200, status)
|
||||
self.assertTrue(as_json(data)["rows"])
|
||||
|
||||
def test_export_csv_forced_to_own_company(self) -> None:
|
||||
status, _, _ = self.cashier_a.get(f"/api/export.csv?company_id={self.company_b}")
|
||||
self.assertEqual(403, status)
|
||||
|
||||
status, headers, data = self.cashier_a.get("/api/export.csv")
|
||||
self.assertEqual(200, status)
|
||||
self.assertEqual("text/csv; charset=utf-8", headers.get("content-type"))
|
||||
text = data.decode("utf-8-sig")
|
||||
lines = [line for line in text.splitlines() if line]
|
||||
self.assertGreater(len(lines), 1)
|
||||
for line in lines[1:]:
|
||||
self.assertEqual(str(self.a_batch_id), line.split(",", 1)[0])
|
||||
|
||||
def test_cross_company_upload_is_recorded_under_own_company(self) -> None:
|
||||
# A uploads B's file bytes while claiming company B in the form; the
|
||||
# server must bind the new (duplicate) batch to A from the session.
|
||||
status, _, data = self.cashier_a.post_multipart(
|
||||
"/api/parse",
|
||||
{"company_id": str(self.company_b)},
|
||||
CCB_SAMPLE.name,
|
||||
CCB_SAMPLE.read_bytes(),
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertEqual("duplicate", as_json(data)["status"])
|
||||
connection = connect(self.db_path)
|
||||
try:
|
||||
duplicate = connection.execute(
|
||||
"SELECT company_id FROM import_batches WHERE status = 'duplicate'"
|
||||
).fetchone()
|
||||
finally:
|
||||
connection.close()
|
||||
self.assertIsNotNone(duplicate)
|
||||
self.assertEqual(self.company_a, duplicate["company_id"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Disable / reset flows
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_disabled_user_session_and_login_rejected(self) -> None:
|
||||
client, password, user_id = self.create_company_user("cashier-disable")
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(200, status)
|
||||
|
||||
status, _, data = self.admin.request("POST", f"/api/admin/users/{user_id}/disable")
|
||||
self.assertEqual(200, status, data)
|
||||
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(401, status)
|
||||
|
||||
fresh = self.fresh_client()
|
||||
status, _, data = fresh.post_json(
|
||||
"/api/login",
|
||||
{"username": "cashier-disable", "password": password, "portal": "company"},
|
||||
)
|
||||
self.assertEqual(403, status, data)
|
||||
|
||||
def test_reset_password_returns_once_and_revokes_sessions(self) -> None:
|
||||
client, old_password, user_id = self.create_company_user("cashier-reset")
|
||||
status, _, data = self.admin.request(
|
||||
"POST", f"/api/admin/users/{user_id}/reset-password"
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
new_password = as_json(data)["initial_password"]
|
||||
self.known_passwords.add(new_password)
|
||||
|
||||
status, _, _ = client.get("/api/me")
|
||||
self.assertEqual(401, status)
|
||||
|
||||
fresh = self.fresh_client()
|
||||
status, _, data = fresh.post_json(
|
||||
"/api/login",
|
||||
{"username": "cashier-reset", "password": new_password, "portal": "company"},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertTrue(as_json(data)["must_change_password"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Secrets hygiene
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_no_response_contains_password_hash(self) -> None:
|
||||
bodies = []
|
||||
status, _, data = self.admin.get("/api/admin/users")
|
||||
self.assertEqual(200, status)
|
||||
bodies.append(data)
|
||||
status, _, data = self.admin.get("/api/admin/audit-log?limit=100")
|
||||
self.assertEqual(200, status)
|
||||
bodies.append(data)
|
||||
status, _, data = self.cashier_a.get("/api/me")
|
||||
bodies.append(data)
|
||||
status, _, data = self.cashier_a.get("/api/batches")
|
||||
bodies.append(data)
|
||||
for body in bodies:
|
||||
self.assertNotIn("password_hash", body.decode("utf-8"))
|
||||
|
||||
def test_audit_log_contains_no_plaintext_passwords(self) -> None:
|
||||
connection = connect(self.db_path)
|
||||
try:
|
||||
rows = connection.execute(
|
||||
"SELECT detail, target FROM audit_log"
|
||||
).fetchall()
|
||||
finally:
|
||||
connection.close()
|
||||
for password in self.known_passwords:
|
||||
for row in rows:
|
||||
self.assertNotIn(password, row["detail"] or "")
|
||||
self.assertNotIn(password, row["target"] or "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user