新增 sys_user 认证现状探查(判断"账号密码登录"可行性)

tools/probe_auth_state.py:只读统计 sys_user 的 status / user_type 分布与 password_hash
的格式特征(只输出前缀与长度,不输出完整值)。

结论(docs/evidence/auth-state.json):库里 5 个用户,password_hash 全是占位符 ——
4 个是 'x'(tools/seed_test_rbac.py 写入),1 个是 '!worker-only-no-password-login!'
(alembic 迁移写入的哨兵值,名字本身就表明"不用于密码登录")。
has_real_password_hash = false。

⇒ 在本项目做"账号密码换令牌"不是"加个端点"的事:没有密码可校验,等于要新建一套密码体系
(选算法 + 加依赖 + 定义写入流程 + 改上游数据),而 password_hash 是基线已有字段
(规则 4:不得改变既有业务含义)。docs/05 §11 也已明确该职责属统一身份认证模块。

顺带记录一个数据质量问题:user_type 取值不统一(employee / 员工 两种写法并存)。
This commit is contained in:
2026-09-11 20:30:27 +08:00
parent 76e87a33a7
commit 62501747a6
2 changed files with 123 additions and 0 deletions
+96
View File
@@ -0,0 +1,96 @@
"""只读探查:`sys_user` 的认证字段现状 —— 判断"账号密码换令牌"这条路通不通。
只做 SELECT;`password_hash` **只统计前缀特征与长度,不输出完整值**(金融项目的基本习惯,
即使是测试库)。
python tools/probe_auth_state.py
"""
from __future__ import annotations
import asyncio
import json
from collections import Counter
from pathlib import Path
from typing import Any
from sqlalchemy import text
from app.infrastructure.db import SessionFactory
OUTPUT = Path("docs/evidence/auth-state.json")
#: 常见密码哈希方案的特征前缀。任何一项命中,说明库里存在**真实可校验**的密码。
KNOWN_HASH_PREFIXES = (
"$2a$", "$2b$", "$2y$", # bcrypt
"$argon2", # argon2
"$pbkdf2", "$pbkdf2-sha256", # passlib pbkdf2
"$5$", "$6$", "$1$", # crypt sha256/sha512/md5
"$scrypt",
)
def _classify(values: list[str]) -> dict[str, Any]:
prefixes: Counter[str] = Counter()
lengths: Counter[int] = Counter()
for value in values:
lengths[len(value)] += 1
matched = next(
(prefix for prefix in KNOWN_HASH_PREFIXES if value.startswith(prefix)), None
)
prefixes[matched or f"<未知格式,首字符 {value[:1]!r}>"] += 1
return {
"length_distribution": dict(sorted(lengths.items())),
"prefix_distribution": dict(prefixes.most_common()),
}
async def collect() -> dict[str, Any]:
report: dict[str, Any] = {}
async with SessionFactory() as session:
rows = (
await session.execute(
text(
"SELECT id, username, user_type, status, password_hash "
"FROM sys_user ORDER BY id"
)
)
).mappings().all()
report["user_count"] = len(rows)
report["by_status"] = dict(
Counter(str(row["status"]) for row in rows).most_common()
)
report["by_user_type"] = dict(
Counter(str(row["user_type"]) for row in rows).most_common()
)
report["password_hash_shape"] = _classify(
[str(row["password_hash"] or "") for row in rows]
)
# 判断"能不能校验密码"的关键结论,直接给出来。
hashes = [str(row["password_hash"] or "") for row in rows]
report["has_real_password_hash"] = any(
value.startswith(KNOWN_HASH_PREFIXES) for value in hashes
)
report["placeholder_examples"] = sorted(
{
value
for value in hashes
if not value.startswith(KNOWN_HASH_PREFIXES)
}
)[:5]
return report
async def main() -> None:
report = await collect()
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
OUTPUT.write_text(
json.dumps(report, ensure_ascii=False, indent=2, default=str),
encoding="utf-8",
)
print(f"wrote {OUTPUT}")
if __name__ == "__main__":
asyncio.run(main())