fix: platform.db 前瞻加固(bootstrap竞态/哈希版本化/json_valid CHECK/测试补空/凭据列警示)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-13 08:08:22 +08:00
parent 2d62108c73
commit bfcaba5c87
2 changed files with 143 additions and 26 deletions
+76 -2
View File
@@ -9,6 +9,7 @@ Windows 纪律:每个建过 engine 的用例结尾 await engine.dispose()——
连接不关,文件锁着,tmp_path 清理 PermissionError。
"""
import asyncio
import hashlib
from pathlib import Path
import pytest
@@ -109,16 +110,43 @@ async def test_platform_db_wal(tmp_path):
# ---- 密码哈希 -----------------------------------------------------------------
def test_hash_password_fresh_salt_per_call():
def test_hash_password_versioned_format_and_fresh_salt():
"""同密码两次哈希:salt/摘要都不同(随机 salt,彩虹表防护);
显式传 salt 可复现(verify_password 内部路径)。"""
显式传 salt 可复现(verify_password 内部路径)。
版本化串(M-1): "scrypt$n$r$p$salt_hex$hash_hex"——参数/盐/摘要
同存,将来提参/换算法有迁移路径。"""
h1, s1 = hash_password("same")
h2, s2 = hash_password("same")
assert s1 != s2 and h1 != h2
parts = h1.split("$")
assert parts[0] == "scrypt" and len(parts) == 6
assert parts[1:4] == ["16384", "8", "1"]
assert parts[4] == s1 # 串内盐与 salt 列(hex)一致
h3, s3 = hash_password("same", salt=bytes.fromhex(s1))
assert h3 == h1 and s3 == s1
def test_verify_password_bad_stored_hash_returns_false():
"""存储异常按验败处理不炸(M-4):坏 hex/坏段数/坏参数 → False;
无前缀过渡行兼容:salt 列 + 默认参数仍可验真(M-1 解析兼容)。"""
bad_stored = [
"not-hex!", # 无前缀,坏 hex
"scrypt$16384$8$1$deadbeef", # 段数不足
"scrypt$abc$8$1$00$00", # 参数非整数
]
for stored in bad_stored:
user = AdminUser(username="u", password_hash=stored, salt="zz")
assert verify_password("whatever", user) is False
salt = bytes(16)
digest = hashlib.scrypt( # 裸摘要 + salt 列的过渡行形态
b"legacy", salt=salt, n=2 ** 14, r=8, p=1)
legacy = AdminUser(
username="u", password_hash=digest.hex(), salt=salt.hex())
assert verify_password("legacy", legacy) is True
assert verify_password("other", legacy) is False
# ---- 模型约束 -------------------------------------------------------------
@@ -158,6 +186,12 @@ async def test_unique_constraints(tmp_path):
await session.commit()
await session.rollback()
session.add(TenantConfigVersion(
tenant_id=t1_id, version=2, config_json="{oops"))
with pytest.raises(IntegrityError): # M-3: json_valid CHECK
await session.commit()
await session.rollback()
session.add(AdminUser(
username="t1admin", password_hash="x", salt="y"))
with pytest.raises(IntegrityError):
@@ -167,6 +201,46 @@ async def test_unique_constraints(tmp_path):
await engine.dispose()
async def test_schema_metadata_isolation(tmp_path):
"""互不误建(M-4):ensure_platform_schema 只建平台表,租户库的
ensure_schema 只建租户表——两套 MetaData 各管各的,engine 指错库
也不会混表。"""
from sqlalchemy import text
from app.engine.models import ensure_schema
from app.platform.registry import make_tenant_engine
platform_tables = {"tenants", "sites", "site_domains",
"tenant_config_versions", "admin_users",
"platform_settings"}
tenant_tables = {"conversations", "messages", "customer_profiles",
"evaluations", "lead_push_logs"}
pe = make_platform_engine(tmp_path / "platform.db")
try:
await ensure_platform_schema(pe)
async with pe.connect() as conn:
names = set((await conn.execute(
text("SELECT name FROM sqlite_master WHERE type='table'")
)).scalars())
assert platform_tables <= names
assert not (tenant_tables & names) # 不建租户表
finally:
await pe.dispose()
te = make_tenant_engine(tmp_path / "tenant.db")
try:
await ensure_schema(te)
async with te.connect() as conn:
names = set((await conn.execute(
text("SELECT name FROM sqlite_master WHERE type='table'")
)).scalars())
assert tenant_tables <= names
assert not (platform_tables & names) # 不建平台表
finally:
await te.dispose()
async def test_delete_tenant_cascades_children(tmp_path):
"""FK 级联(make_platform_engine 每连接 PRAGMA foreign_keys=ON):
删租户 → sites/site_domains/tenant_config_versions/该租户的 admin_users