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())