diff --git a/app/platform/db.py b/app/platform/db.py index b826f2b..eb3b4a7 100644 --- a/app/platform/db.py +++ b/app/platform/db.py @@ -14,10 +14,13 @@ PlatformBase.metadata,engine/models.ensure_schema 只跑 Base.metadata—— 旧版复制成新版本,Task 8);UNIQUE(tenant_id, version)。config_json 用 Text 存 JSON 字符串,读写方显式 json.loads/dumps(快照几十 KB,保持 直存直取,不经 ORM JSON 类型的隐式序列化)。 -- admin_users:tenant_id NULL = 超管。password_hash/salt 存 hex 文本; - hash = hashlib.scrypt(n=2**14, r=8, p=1)——偏差注记:spec §7.2 写 - bcrypt/argon2,用 stdlib scrypt 为有意取舍(零新依赖,抗 GPU 性能同级); - verify 走 hmac.compare_digest 常量时间比较。 +- admin_users:tenant_id NULL = 超管。password_hash 存版本化串 + "scrypt$$$

$$"(参数与摘要同存,将来提参/ + 换算法有迁移路径,verify 解析兼容);salt 列同步存同一盐 hex(单一写方 + hash_password,verify 以串内为准)。哈希 = hashlib.scrypt(n=2**14, r=8, + p=1)——偏差注记:spec §7.2 写 bcrypt/argon2,用 stdlib scrypt 为有意 + 取舍(零新依赖,抗 GPU 性能同级);verify 走 hmac.compare_digest + 常量时间比较。 测试/teardown 纪律:建过 platform engine 的测试结尾 await engine.dispose() (Windows 文件锁,同 registry)。 @@ -31,6 +34,7 @@ import secrets from sqlalchemy import ( JSON, Boolean, + CheckConstraint, ForeignKey, Integer, String, @@ -39,6 +43,7 @@ from sqlalchemy import ( event, select, ) +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, @@ -96,10 +101,16 @@ class SiteDomain(PlatformBase): class TenantConfigVersion(PlatformBase): """租户配置版本:每次保存存整份快照(几十 KB,不做增量); - 当前生效版 = MAX(version);UNIQUE(tenant_id, version)。""" + 当前生效版 = MAX(version);UNIQUE(tenant_id, version)。 + CHECK(json_valid(config_json)):坏 JSON 在库层兜底拒收(当前零写入方, + 加约束零破坏;Task 5+ 写入方还应前置 json.dumps——约束只兜底不代偿)。""" __tablename__ = "tenant_config_versions" - __table_args__ = (UniqueConstraint("tenant_id", "version"),) + __table_args__ = ( + UniqueConstraint("tenant_id", "version"), + CheckConstraint( + "json_valid(config_json)", name="ck_tcv_config_json_valid"), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True) tenant_id: Mapped[int] = mapped_column( @@ -170,27 +181,49 @@ _SCRYPT_P = 1 def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]: - """scrypt(password, salt) → (password_hash, salt),均 hex 文本(便于 - 直读排查);salt 缺省每次随机 16 字节。显式传 salt 可复现(测试/verify)。""" + """scrypt(password, salt) → (password_hash, salt)。 + + password_hash 为版本化串 "scrypt$$$

$$": + 参数与摘要同存,将来提参/换算法时旧行自解释,有迁移路径(M-1)。 + salt 返回值(列存)与串内嵌的同一盐 hex——单一写方在此同步两处, + verify 以串内为准。salt 缺省每次随机 16 字节;显式传 salt 可复现 + (测试/verify)。""" if salt is None: salt = secrets.token_bytes(16) digest = hashlib.scrypt( password.encode("utf-8"), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P) - return digest.hex(), salt.hex() + versioned = ( + f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}" + f"${salt.hex()}${digest.hex()}") + return versioned, salt.hex() def verify_password(password: str, user: AdminUser) -> bool: - """常量时间比较(hmac.compare_digest)——防时序侧信道;坏 hex 按验败 - 处理而非炸(存储异常不该变成 500)。""" + """解析 password_hash 版本化串重算比对,常量时间比较 + (hmac.compare_digest)——防时序侧信道。 + + 解析兼容(M-1):带 "scrypt$" 前缀 → 参数/盐/摘要全取自串内; + 无前缀(过渡/手工行)→ 盐取 salt 列、参数取当前默认。任何解析失败 + (坏 hex/坏段数/坏整数)按验败返回 False 而非炸——存储异常不该变 500。""" + stored = user.password_hash try: - salt = bytes.fromhex(user.salt) - expected = bytes.fromhex(user.password_hash) + if stored.startswith("scrypt$"): + parts = stored.split("$") + if len(parts) != 6: + return False + _, n_s, r_s, p_s, salt_hex, hash_hex = parts + n, r, p = int(n_s), int(r_s), int(p_s) + salt = bytes.fromhex(salt_hex) + expected = bytes.fromhex(hash_hex) + else: + salt = bytes.fromhex(user.salt) + expected = bytes.fromhex(stored) + n, r, p = _SCRYPT_N, _SCRYPT_R, _SCRYPT_P except ValueError: return False actual = hashlib.scrypt( - password.encode("utf-8"), salt=salt, - n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P) + password.encode("utf-8"), salt=salt, n=n, r=r, p=p) return hmac.compare_digest(expected, actual) @@ -200,24 +233,34 @@ async def bootstrap_superuser( engine: AsyncEngine, username: str, password: str) -> None: """空表才建超管(tenant_id NULL, must_change_password=True);表非空 一律不动——不重复建、不改已有用户/密码(幂等:lifespan 每次启动安全 - 重入,运行期改密不会被启动覆盖)。""" + 重入,运行期改密不会被启动覆盖)。 + + 多进程共享库竞态(I-1):两进程同时见空表同时插,UNIQUE(username) + 让输方 IntegrityError——吞掉视为已引导(rollback 后返回),输方启动 + 不炸,库落一行赢家结果,语义等价。""" async with AsyncSession(engine, expire_on_commit=False) as session: existing = (await session.execute( select(AdminUser.id).limit(1))).first() if existing is not None: return password_hash, salt = hash_password(password) - session.add(AdminUser( - tenant_id=None, - username=username, - password_hash=password_hash, - salt=salt, - must_change_password=True)) - await session.commit() + try: + session.add(AdminUser( + tenant_id=None, + username=username, + password_hash=password_hash, + salt=salt, + must_change_password=True)) + await session.commit() + except IntegrityError: + await session.rollback() # 输方:另一进程已抢先建,视为已引导 async def list_admin_users(engine: AsyncEngine) -> list[AdminUser]: - """全部后台账号(按 id 稳序)——测试/管理列表用。""" + """全部后台账号(按 id 稳序)——测试/管理列表用。 + + 警示(M-5):返回的是 ORM 实体,含凭据列(password_hash/salt)—— + 序列化出 API/日志前必须显式剔除这两列,凭据永不外泄。""" async with AsyncSession(engine, expire_on_commit=False) as session: result = await session.execute(select(AdminUser).order_by(AdminUser.id)) return list(result.scalars().all()) diff --git a/tests/test_platform_db.py b/tests/test_platform_db.py index 885f8ac..58bc5f0 100644 --- a/tests/test_platform_db.py +++ b/tests/test_platform_db.py @@ -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