From 62501747a6c4de22b08e3c1b01b5802943a88daa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 20:30:27 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=20sys=5Fuser=20=E8=AE=A4?= =?UTF-8?q?=E8=AF=81=E7=8E=B0=E7=8A=B6=E6=8E=A2=E6=9F=A5=EF=BC=88=E5=88=A4?= =?UTF-8?q?=E6=96=AD"=E8=B4=A6=E5=8F=B7=E5=AF=86=E7=A0=81=E7=99=BB?= =?UTF-8?q?=E5=BD=95"=E5=8F=AF=E8=A1=8C=E6=80=A7=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 / 员工 两种写法并存)。 --- docs/evidence/auth-state.json | 27 ++++++++++ tools/probe_auth_state.py | 96 +++++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+) create mode 100644 docs/evidence/auth-state.json create mode 100644 tools/probe_auth_state.py diff --git a/docs/evidence/auth-state.json b/docs/evidence/auth-state.json new file mode 100644 index 0000000..ac35093 --- /dev/null +++ b/docs/evidence/auth-state.json @@ -0,0 +1,27 @@ +{ + "user_count": 5, + "by_status": { + "正常": 4, + "禁用": 1 + }, + "by_user_type": { + "employee": 3, + "customer": 1, + "员工": 1 + }, + "password_hash_shape": { + "length_distribution": { + "1": 4, + "31": 1 + }, + "prefix_distribution": { + "<未知格式,首字符 'x'>": 4, + "<未知格式,首字符 '!'>": 1 + } + }, + "has_real_password_hash": false, + "placeholder_examples": [ + "!worker-only-no-password-login!", + "x" + ] +} \ No newline at end of file diff --git a/tools/probe_auth_state.py b/tools/probe_auth_state.py new file mode 100644 index 0000000..6b22e04 --- /dev/null +++ b/tools/probe_auth_state.py @@ -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())